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

N = 1
observations = []
def solve(events):
    order=[]; state={}; out=[]
    for e in events:
        if e[0]=='open':
            order=list(e[1]); state={k:['pending',None] for k in order}
        elif e[0]=='result':
            k,status,value=e[1:]
            if k not in state: continue
            if state[k][0] in ('success','permanent'): continue
            if status=='success': state[k]=['success',value]
            elif status=='permanent': state[k]=['permanent',None]
            else: state[k]=['retry',None]
        elif e[0]=='retry':
            out.append([k for k in order if state[k][0] in ('pending','retry')])
        elif e[0]=='done':
            out.append([state[k][1] if state[k][0]=='success' else None for k in sorted(order)])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('open',['b','a','b']),('retry',),('result','b','success',N),('done',),('retry',)]), [['b','a','b'],[N,None,N],['a']])
check('2', solve([('open',['a','b']),('result','a','retry',N),('result','b','permanent',N),('retry',),('done',)]), [['a'],[None,None]])
check('3', solve([('open',['a']),('result','z','success',N),('retry',),('done',)]), [['a'],[None]])
check('4', solve([('open',['a','b']),('result','a','success',N),('result','a','retry',0),('result','b','permanent',0),('result','b','retry',0),('retry',),('done',)]), [[],[N,None]])
check('5', solve([('open',['a']),('result','a','success',N),('open',['a']),('retry',),('done',)]), [['a'],[None]])
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):
    order=[]; state={}; out=[]
    for e in events:
        if e[0]=='open':
            order=list(e[1]); state={k:['pending',None] for k in order}
        elif e[0]=='result':
            k,status,value=e[1:]
            if k not in state: continue
            if state[k][0] in ('success','permanent'): continue
            if status=='success': state[k]=['success',value]
            elif status=='permanent': state[k]=['permanent',None]
            else: state[k]=['retry',None]
        elif e[0]=='retry':
            out.append([k for k in order if state[k][0] in ('pending','retry')])
        elif e[0]=='done':
            out.append([state[k][1] if state[k][0]=='success' else None for k in dict.fromkeys(order)])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('open',['b','a','b']),('retry',),('result','b','success',N),('done',),('retry',)]), [['b','a','b'],[N,None,N],['a']])
check('2', solve([('open',['a','b']),('result','a','retry',N),('result','b','permanent',N),('retry',),('done',)]), [['a'],[None,None]])
check('3', solve([('open',['a']),('result','z','success',N),('retry',),('done',)]), [['a'],[None]])
check('4', solve([('open',['a','b']),('result','a','success',N),('result','a','retry',0),('result','b','permanent',0),('result','b','retry',0),('retry',),('done',)]), [[],[N,None]])
check('5', solve([('open',['a']),('result','a','success',N),('open',['a']),('retry',),('done',)]), [['a'],[None]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Final batch response loses original position mapping

open(list of item IDs) establishes ordered batch. result(id,status,value) records first success, retryable failure, or permanent failure; unknown IDs ignored. retry returns unresolved/retryable items in original order including duplicates. done returns values or null in original positions. Success/permanent terminal entries cannot be overwritten by late responses.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if False: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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)

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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal and delivered>0: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Retry starts after an empty response already completed

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=False
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=delivered>0
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
An initial response attempt is not marked active

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=200
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=status
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
A retry transport failure inherits the preceding response status

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=200
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=None
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
Partial response errors lose the actual HTTP response status

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=1
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
Zero-length callbacks erase or invent consumer-visible delivery

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if False: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append(['retry',status]); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
Repeated failure callbacks authorize retries without an active attempt

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=True
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=delivered==0
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Failure leaves the response attempt active for duplicate retry admission

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>1:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>=0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],0])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Retry suppression mishandles one-byte or empty consumer delivery

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=False
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=delivered>1
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=True; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
A partially exposed response is not terminally sealed

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=False; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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):
    active=False; delivered=0; status=None; terminal=False; out=[]
    for e in events:
        if e[0]=='begin':
            if terminal: out.append('terminal'); continue
            active=True
            status=None
        elif e[0]=='headers' and active:
            status=e[1]
        elif e[0]=='deliver' and active:
            delivered+=e[1]
        elif e[0]=='fail':
            if terminal: out.append('terminal'); continue
            if not active: out.append('inactive'); continue
            active=False
            if delivered>0:
                terminal=True
                out.append(['partial',status])
            else: out.append(['retry',status])
        elif e[0]=='finish' and active:
            terminal=delivered>0; active=False; out.append('success')
        elif e[0]=='cancel':
            terminal=True; active=False; out.append('cancelled')
    return [out,delivered]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('fail',)]), [['inactive'],0])
check('2', solve([('begin',),('headers',503),('deliver',0),('fail',),('fail',),('begin',),('fail',)]), [[['retry',503],'inactive',['retry',None]],0])
check('3', solve([('begin',),('headers',200),('deliver',1),('deliver',0),('fail',),('begin',)]), [[['partial',200],'terminal'],1])
check('4', solve([('begin',),('headers',502),('deliver',N+2),('deliver',2),('fail',)]), [[['partial',502]],N+4])
check('5', solve([('begin',),('finish',),('begin',),('fail',)]), [['success','terminal','terminal'],0])
check('6', solve([('begin',),('cancel',),('begin',)]), [['cancelled','terminal'],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:
Completing an empty response does not stop later attempts

begin starts an attempt only while nonterminal, headers records a status, deliver(bytes) records irreversible delivery, fail allows retry only before any delivered bytes. A zero-byte delivery is reversible. finish terminally succeeds. cancel terminally cancels. New attempts reset their local status but never erase already exposed data. Return retry decisions and exposed byte count.

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None and boundary>0: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Block draining violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[0]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]//2]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Stream identity violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],False]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],bool(e[1]%2==0)]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Initial active violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else max(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else e[1]
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Goaway min violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=0 if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1]+1 if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
First boundary violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=True
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=e[1]>0
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Success closes violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('retry'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('ambiguous'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Unknown result violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('retry'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('ambiguous'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Finished result violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=True
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=boundary is not None
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Consume failure violates retry transition semantics

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

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>=boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]<boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Processed boundary violates retry transition semantics

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

split: open-access