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

N = 1
observations = []
def solve(events):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current[1:]!=snapshot[1:]:
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current[1]!=snapshot[1]:
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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:
Identity check violates retry transition semantics

open(file identity,size,mtime,offset) snapshots file metadata and starting byte. inspect(current identity,size,mtime) records current metadata. read(count) advances bounded cursor only while valid. retry compares all snapshot metadata before rewinding; any mismatch permanently invalidates this body. close invalidates. New open resets validity. Return read byte counts and retry dispositions.

split: open-access


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

N = 1
observations = []
def solve(events):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current[:2]!=snapshot[:2]:
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current[0]!=snapshot[0]:
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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:
Mtime check violates retry transition semantics

open(file identity,size,mtime,offset) snapshots file metadata and starting byte. inspect(current identity,size,mtime) records current metadata. read(count) advances bounded cursor only while valid. retry compares all snapshot metadata before rewinding; any mismatch permanently invalidates this body. close invalidates. New open resets validity. Return read byte counts and retry dispositions.

split: open-access


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

N = 1
observations = []
def solve(events):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current!=snapshot:
                valid=True; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current!=snapshot:
                valid=current[0]==snapshot[0]; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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:
Invalidate changed violates retry transition semantics

open(file identity,size,mtime,offset) snapshots file metadata and starting byte. inspect(current identity,size,mtime) records current metadata. read(count) advances bounded cursor only while valid. retry compares all snapshot metadata before rewinding; any mismatch permanently invalidates this body. close invalidates. New open resets validity. Return read byte counts and retry dispositions.

split: open-access


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

N = 1
observations = []
def solve(events):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if (current[0],current[2])!=(snapshot[0],snapshot[2]):
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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):
    snapshot=None; current=None; origin=0; cursor=0; valid=False; out=[]
    for e in events:
        if e[0]=='open':
            snapshot=tuple(e[1:4]); current=snapshot
            origin=e[4]; cursor=origin
            valid=True
        elif e[0]=='inspect': current=tuple(e[1:4])
        elif e[0]=='close': valid=False
        elif e[0]=='read':
            if not valid: out.append('invalid'); continue
            amount=min(e[1],max(0,snapshot[1]-cursor))
            cursor+=amount
            out.append(amount)
        elif e[0]=='retry':
            if not valid: out.append('invalid'); continue
            if current[2]!=snapshot[2]:
                valid=False; out.append('changed'); continue
            cursor=origin
            out.append('rewound')
    return [out,cursor]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('open','a',N+5,1,2),('read',2),('read',N+9),('read',1),('retry',),('read',1)]), [[2,N+1,0,'rewound',1],3])
check('2', solve([('open','a',5,1,0),('inspect','b',5,1),('retry',),('read',1)]), [['changed','invalid'],0])
check('3', solve([('open','a',5,1,0),('inspect','a',6,1),('retry',),('retry',)]), [['changed','invalid'],0])
check('4', solve([('open','a',5,1,0),('inspect','a',5,2),('retry',),('inspect','a',5,1),('retry',)]), [['changed','invalid'],0])
check('5', solve([('open','a',0,1,0),('read',1),('retry',),('close',),('read',1),('retry',)]), [[0,'rewound','invalid','invalid'],0])
check('6', solve([('open','a',N+3,1,1),('close',),('open','b',2,2,1),('read',1)]), [[1],2])
check('7', solve([('open','a',N+3,1,1),('inspect','a',N+3,1),('retry',),('close',),('read',1),('retry',)]), [['rewound','invalid','invalid'],1])
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:
Size check violates retry transition semantics

open(file identity,size,mtime,offset) snapshots file metadata and starting byte. inspect(current identity,size,mtime) records current metadata. read(count) advances bounded cursor only while valid. retry compares all snapshot metadata before rewinding; any mismatch permanently invalidates this body. close invalidates. New open resets validity. Return read byte counts and retry dispositions.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=0; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=max(0,e[1]-1); closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Remaining length violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=True; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=e[1]==0; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Response open violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=True
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=eof
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Eof reset violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append(0); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed and remaining==0: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Read closed violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=e[1]; remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=remaining; remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Drain bound violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining=0; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining=max(0,remaining-1); out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Drain accounting violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=False
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining>1: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Early eof poison violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=False
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=closed
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Eof confirm violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=False
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=remaining==0
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Close state violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=False; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=remaining>0; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Abort poison violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('pooled')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison and remaining>0: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Poison first violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('pooled')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed and remaining>0: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Release busy violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining<0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining>1: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Release unread violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif False: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=False
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof and remaining>0: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Release framing violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=poison
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    remaining=0; closed=True; eof=False; poison=False; out=[]
    for e in events:
        if e[0]=='new': remaining=0; closed=True; eof=False; poison=True
        elif e[0]=='headers':
            remaining=e[1]; closed=False; eof=False
        elif e[0]=='read':
            if closed: out.append('closed'); continue
            n=min(remaining,e[1]); remaining-=n; out.append(n)
        elif e[0]=='eof':
            if remaining!=0: poison=True
            else: eof=True
        elif e[0]=='close': closed=True
        elif e[0]=='abort': poison=True; closed=True
        elif e[0]=='release':
            if poison: out.append('discard')
            elif not closed: out.append('busy')
            elif remaining!=0: out.append('unread')
            elif not eof: out.append('unframed')
            else: out.append('pooled')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('headers',N+3),('read',2),('read',N+9),('eof',),('release',),('close',),('release',)]), [2,N+1,'busy','pooled'])
check('2', solve([('headers',1),('close',),('read',1),('release',)]), ['closed','unread'])
check('3', solve([('headers',1),('eof',),('read',1),('close',),('release',)]), [1,'discard'])
check('4', solve([('headers',0),('close',),('release',),('eof',),('release',)]), ['unframed','pooled'])
check('5', solve([('headers',0),('eof',),('abort',),('release',),('new',),('headers',0),('eof',),('close',),('release',)]), ['discard','pooled'])
check('6', solve([('headers',0),('eof',),('close',),('headers',0),('close',),('release',)]), ['unframed'])
check('7', solve([('headers',0),('read',1)]), [0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
New clears poison violates retry transition semantics

headers(length) opens failed response with remaining bytes; read(count) drains bounded content. eof confirms framing only if no bytes remain, otherwise poisons connection. close marks response closed; release returns pooled only for closed, fully drained, EOF-confirmed unpoisoned response. abort permanently poisons current connection. new resets all connection state. Return drain amounts and release disposition.

split: open-access


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

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

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

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

prompt:
Configure cap violates retry transition semantics

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

split: open-access


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

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

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

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

prompt:
Configure statuses violates retry transition semantics

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

split: open-access