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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].clear()
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-pop fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('request-pop fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('request-pop fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('request-pop fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('request-pop fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('request-pop fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('request-pop fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('request-pop fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop()
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-pop fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('request-pop fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('request-pop fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('request-pop fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('request-pop fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('request-pop fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('request-pop fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('request-pop fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent download clients share bounded origin range reads: only the completed front request is removed from a consumer queue.

x contains ordered jobs [client,absolute-start,count,cancelled], positive chunk size, and dispatch slot count. Drop canceled and zero-length jobs. Maintain first-appearance client order; each client has FIFO requests. Dispatch at most one chunk per client per round, rotate unfinished clients, split long ranges, and remove exhausted requests. Return [dispatches,pending-jobs], where dispatches are [client,start,count] and pending jobs retain scheduler order. This models range-specific origin request scheduling, without network retries.

split: open-access


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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-origin fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('remaining-origin fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('remaining-origin fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('remaining-origin fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('remaining-origin fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('remaining-origin fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('remaining-origin fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('remaining-origin fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[amount,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-origin fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('remaining-origin fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('remaining-origin fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('remaining-origin fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('remaining-origin fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('remaining-origin fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('remaining-origin fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('remaining-origin fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent download clients share bounded origin range reads: subsequent chunks advance their absolute origin byte start.

x contains ordered jobs [client,absolute-start,count,cancelled], positive chunk size, and dispatch slot count. Drop canceled and zero-length jobs. Maintain first-appearance client order; each client has FIFO requests. Dispatch at most one chunk per client per round, rotate unfinished clients, split long ranges, and remove exhausted requests. Return [dispatches,pending-jobs], where dispatches are [client,start,count] and pending jobs retain scheduler order. This models range-specific origin request scheduling, without network retries.

split: open-access


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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,count]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-count fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('remaining-count fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('remaining-count fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('remaining-count fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('remaining-count fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('remaining-count fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('remaining-count fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('remaining-count fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-count fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('remaining-count fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('remaining-count fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('remaining-count fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('remaining-count fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('remaining-count fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('remaining-count fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('remaining-count fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent download clients share bounded origin range reads: scheduling consumes only the dispatched chunk from the request.

x contains ordered jobs [client,absolute-start,count,cancelled], positive chunk size, and dispatch slot count. Drop canceled and zero-length jobs. Maintain first-appearance client order; each client has FIFO requests. Dispatch at most one chunk per client per round, rotate unfinished clients, split long ranges, and remove exhausted requests. Return [dispatches,pending-jobs], where dispatches are [client,start,count] and pending jobs retain scheduler order. This models range-specific origin request scheduling, without network retries.

split: open-access


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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client]: order.insert(0,client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('round-robin-rotation fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('round-robin-rotation fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('round-robin-rotation fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('round-robin-rotation fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('round-robin-rotation fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('round-robin-rotation fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('round-robin-rotation fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('round-robin-rotation fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client] and len(queues[client])>1: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('round-robin-rotation fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('round-robin-rotation fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('round-robin-rotation fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('round-robin-rotation fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('round-robin-rotation fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('round-robin-rotation fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('round-robin-rotation fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('round-robin-rotation fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent download clients share bounded origin range reads: an unfinished large range yields the origin slot to other clients.

x contains ordered jobs [client,absolute-start,count,cancelled], positive chunk size, and dispatch slot count. Drop canceled and zero-length jobs. Maintain first-appearance client order; each client has FIFO requests. Dispatch at most one chunk per client per round, rotate unfinished clients, split long ranges, and remove exhausted requests. Return [dispatches,pending-jobs], where dispatches are [client,start,count] and pending jobs retain scheduler order. This models range-specific origin request scheduling, without network retries.

split: open-access


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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client][:1]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('pending-request-retention fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('pending-request-retention fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('pending-request-retention fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('pending-request-retention fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('pending-request-retention fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('pending-request-retention fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('pending-request-retention fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('pending-request-retention fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    queues={}
    order=[]
    for client,start,count,cancelled in x['jobs']:
        if cancelled or count==0: continue
        if client not in queues:
            queues[client]=[]
            order.append(client)
        queues[client].append([start,count])
    out=[]
    for slot in range(x['slots']):
        if not order: break
        client=order.pop(0)
        start,count=queues[client][0]
        amount=min(count,x['chunk'])
        out.append([client,start,amount])
        if amount==count:
            queues[client].pop(0)
        else:
            queues[client][0]=[start+amount,count-amount]
        if queues[client]: order.append(client)
    pending=[[client,start,count] for client in order for start,count in queues[client][-1:]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('pending-request-retention fixture 0', json.loads(json.dumps(solve({'jobs':[['a',0,5,False],['b',10,2,False]],'chunk':2,'slots':4}))), json.loads(json.dumps([[['a',0,2],['b',10,2],['a',2,2],['a',4,1]],[]])))
check('pending-request-retention fixture 1', json.loads(json.dumps(solve({'jobs':[['b',10,2,False],['a',0,2,False]],'chunk':1,'slots':2}))), json.loads(json.dumps([[['b',10,1],['a',0,1]],[['b',11,1],['a',1,1]]])))
check('pending-request-retention fixture 2', json.loads(json.dumps(solve({'jobs':[['a',0,0,False],['b',10,2,True]],'chunk':1,'slots':2}))), json.loads(json.dumps([[],[]])))
check('pending-request-retention fixture 3', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',9,1,False],['b',4,1,False]],'chunk':2,'slots':2}))), json.loads(json.dumps([[['a',0,2],['b',4,1]],[['a',9,1]]])))
check('pending-request-retention fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('pending-request-retention fixture 5', json.loads(json.dumps(solve({'jobs':[['a',0,2,False],['a',8,1,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2],['a',8,1]]])))
check('pending-request-retention fixture 6', json.loads(json.dumps(solve({'jobs':[['a',N,N+1,False]],'chunk':N,'slots':1}))), json.loads(json.dumps([[['a',N,N]],[['a',2*N,1]]])))
check('pending-request-retention fixture 7', json.loads(json.dumps(solve({'jobs':[['a',0,1,False]],'chunk':3,'slots':2}))), json.loads(json.dumps([[['a',0,1]],[]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent download clients share bounded origin range reads: undispatched requests from the same consumer survive the scheduling turn.

x contains ordered jobs [client,absolute-start,count,cancelled], positive chunk size, and dispatch slot count. Drop canceled and zero-length jobs. Maintain first-appearance client order; each client has FIFO requests. Dispatch at most one chunk per client per round, rotate unfinished clients, split long ranges, and remove exhausted requests. Return [dispatches,pending-jobs], where dispatches are [client,start,count] and pending jobs retain scheduler order. This models range-specific origin request scheduling, without network retries.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=(cursor+block-1)//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('storage-page fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('storage-page fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('storage-page fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('storage-page fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('storage-page fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('storage-page fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('storage-page fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=round(cursor/block)
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('storage-page fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('storage-page fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('storage-page fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('storage-page fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('storage-page fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('storage-page fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('storage-page fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: byte positions map by floor division into storage blocks.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=0
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('intra-page-offset fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('intra-page-offset fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('intra-page-offset fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('intra-page-offset fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('intra-page-offset fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('intra-page-offset fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('intra-page-offset fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor%block if cursor>=block else 0
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('intra-page-offset fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('intra-page-offset fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('intra-page-offset fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('intra-page-offset fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('intra-page-offset fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('intra-page-offset fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('intra-page-offset fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: slices retain their offset inside the first aligned block.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('page-tail fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('page-tail fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('page-tail fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('page-tail fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('page-tail fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('page-tail fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('page-tail fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,max(1,block-local-1))
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('page-tail fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('page-tail fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('page-tail fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('page-tail fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('page-tail fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('page-tail fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('page-tail fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: each slice stops at the current storage block boundary.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts=[[page,local,count]]
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-map-append fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('request-map-append fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('request-map-append fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('request-map-append fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-append fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-append fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('request-map-append fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.insert(0,[page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-map-append fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('request-map-append fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('request-map-append fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('request-map-append fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-append fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-append fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('request-map-append fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: one request keeps all of its crossed block slices.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            if local==0: needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('fetch-page-set fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('fetch-page-set fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('fetch-page-set fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('fetch-page-set fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('fetch-page-set fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('fetch-page-set fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('fetch-page-set fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            if count==block: needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('fetch-page-set fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('fetch-page-set fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('fetch-page-set fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('fetch-page-set fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('fetch-page-set fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('fetch-page-set fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('fetch-page-set fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: every crossed block contributes to the origin fetch set.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=block
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cursor-step fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('cursor-step fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('cursor-step fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('cursor-step fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cursor-step fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cursor-step fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('cursor-step fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=max(1,count-1)
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cursor-step fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('cursor-step fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('cursor-step fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('cursor-step fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cursor-step fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cursor-step fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('cursor-step fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: the next block slice begins after bytes actually mapped.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        if parts: maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-map-order fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('request-map-order fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('request-map-order fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('request-map-order fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-order fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-order fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('request-map-order fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.insert(0,parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('request-map-order fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('request-map-order fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('request-map-order fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('request-map-order fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-order fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('request-map-order fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('request-map-order fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: empty byte requests retain their location in the response plan.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if False: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cached-block-fetch fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('cached-block-fetch fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('cached-block-fetch fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('cached-block-fetch fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cached-block-fetch fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cached-block-fetch fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('cached-block-fetch fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page not in x['cached']: continue
        absolute=page*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cached-block-fetch fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('cached-block-fetch fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('cached-block-fetch fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('cached-block-fetch fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cached-block-fetch fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('cached-block-fetch fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('cached-block-fetch fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: cached blocks retain slice mappings while skipping origin reads.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('aligned-absolute fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('aligned-absolute fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('aligned-absolute fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('aligned-absolute fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('aligned-absolute fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('aligned-absolute fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('aligned-absolute fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=(page+1)*block
        count=min(block,x['length']-absolute)
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('aligned-absolute fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('aligned-absolute fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('aligned-absolute fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('aligned-absolute fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('aligned-absolute fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('aligned-absolute fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('aligned-absolute fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: physical reads begin at the block base in the object.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=block
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('eof-read-cap fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('eof-read-cap fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('eof-read-cap fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('eof-read-cap fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('eof-read-cap fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('eof-read-cap fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('eof-read-cap fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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(x):
    block=x['block']
    needed=set()
    maps=[]
    for start,stop in x['requests']:
        parts=[]
        cursor=start
        while cursor<stop:
            page=cursor//block
            local=cursor-page*block
            count=min(stop-cursor,block-local)
            parts.append([page,local,count])
            needed.add(page)
            cursor+=count
        maps.append(parts)
    reads=[]
    for page in sorted(needed):
        if page in x['cached']: continue
        absolute=page*block
        count=max(0,min(block,x['length']-absolute-1))
        reads.append([page,absolute,count])
    return [reads,maps]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('eof-read-cap fixture 0', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[3,9]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4],[2,8,2]],[[[0,3,1],[1,0,4],[2,0,1]]]])))
check('eof-read-cap fixture 1', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[8,10],[0,2],[1,3]],'cached':[0]}))), json.loads(json.dumps([[[2,8,2]],[[[2,0,2]],[[0,0,2]],[[0,1,2]]]])))
check('eof-read-cap fixture 2', json.loads(json.dumps(solve({'length':8,'block':4,'requests':[[0,4],[4,8]],'cached':[]}))), json.loads(json.dumps([[[0,0,4],[1,4,4]],[[[0,0,4]],[[1,0,4]]]])))
check('eof-read-cap fixture 3', json.loads(json.dumps(solve({'length':0,'block':4,'requests':[[0,0]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('eof-read-cap fixture 4', json.loads(json.dumps(solve({'length':10,'block':4,'requests':[[2,2]],'cached':[]}))), json.loads(json.dumps([[],[[]]])))
check('eof-read-cap fixture 5', json.loads(json.dumps(solve({'length':3*N,'block':N,'requests':[[0,N],[N,2*N]],'cached':[0]}))), json.loads(json.dumps([[[1,N,N]],[[[0,0,N]],[[1,0,N]]]])))
check('eof-read-cap fixture 6', json.loads(json.dumps(solve({'length':5,'block':4,'requests':[],'cached':[0]}))), json.loads(json.dumps([[],[]])))
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 origin adapter turns byte requests into bounded aligned storage reads: the final storage block does not read beyond representation EOF.

x contains complete object length, positive storage block size, ordered [start,stop) requested spans already within object, and cached block numbers. Return [uncached-block-reads,request-to-block-slices]. Each storage read is [block,absolute-start,count], clipped at EOF. Request maps contain [block,offset,count] for every crossed block, including cached ones; empty spans map to an empty list. Storage reads are deduplicated and sorted by block.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['wire']
    pos=0
    body=''
    trailers=[]
    while True:
        end=data.find('\r\n',pos)
        if end<0: return None
        digits=data[pos:end].split(';',1)[0]
        if not 1<=len(digits)<=2 or any(c not in '0123456789abcdefABCDEF' for c in digits): return None
        size=int(digits,10) if digits.isdecimal() else 0
        pos=end+2
        if size==0:
            while True:
                end=data.find('\r\n',pos)
                if end<0: return None
                line=data[pos:end]
                pos=end+2
                if not line: break
                if ':' not in line: return None
                trailers.append(line)
            if pos!=len(data): return None
            return [body[x['start']:x['stop']],len(body),trailers]
        if pos+size+2>len(data): return None
        if data[pos+size:pos+size+2]!='\r\n': return None
        body+=data[pos:pos+size]
        if len(body)>x['limit']: return None
        pos+=size+2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('hexadecimal-size fixture 0', json.loads(json.dumps(solve({'wire':'A;foo=bar\r\n0123456789\r\n0\r\nX:y\r\n\r\n','start':2,'stop':6,'limit':20}))), json.loads(json.dumps(['2345',10,['X:y']])))
check('hexadecimal-size fixture 1', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n3\r\ncde\r\n0\r\n\r\n','start':1,'stop':4,'limit':5}))), json.loads(json.dumps(['bcd',5,[]])))
check('hexadecimal-size fixture 2', json.loads(json.dumps(solve({'wire':'0\r\n\r\n','start':0,'stop':3,'limit':0}))), json.loads(json.dumps(['',0,[]])))
check('hexadecimal-size fixture 3', json.loads(json.dumps(solve({'wire':'1;z=q\r\nx\r\n0\r\n\r\n','start':0,'stop':1,'limit':1}))), json.loads(json.dumps(['x',1,[]])))
check('hexadecimal-size fixture 4', json.loads(json.dumps(solve({'wire':'1\r\nxXX0\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 5', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\n\r\nextra','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 6', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nBadTrailer\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 7', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n2\r\ncd\r\n0\r\n\r\n','start':0,'stop':4,'limit':3}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 8', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nX:a\r\nY:b\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(['x',1,['X:a','Y:b']])))
check('hexadecimal-size fixture 9', json.loads(json.dumps(solve({'wire':format(N,"x")+"\r\n"+"p"*N+"\r\n0\r\n\r\n",'start':0,'stop':N,'limit':10}))), json.loads(json.dumps(["p"*N,N,[]])))
check('hexadecimal-size fixture 10', json.loads(json.dumps(solve({'wire':'10\r\nabcdefghijklmnop\r\n0\r\n\r\n','start':12,'stop':16,'limit':16}))), json.loads(json.dumps(['mnop',16,[]])))
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(x):
    data=x['wire']
    pos=0
    body=''
    trailers=[]
    while True:
        end=data.find('\r\n',pos)
        if end<0: return None
        digits=data[pos:end].split(';',1)[0]
        if not 1<=len(digits)<=2 or any(c not in '0123456789abcdefABCDEF' for c in digits): return None
        size=int(digits,16) if not digits.isdecimal() else int(digits,10)
        pos=end+2
        if size==0:
            while True:
                end=data.find('\r\n',pos)
                if end<0: return None
                line=data[pos:end]
                pos=end+2
                if not line: break
                if ':' not in line: return None
                trailers.append(line)
            if pos!=len(data): return None
            return [body[x['start']:x['stop']],len(body),trailers]
        if pos+size+2>len(data): return None
        if data[pos+size:pos+size+2]!='\r\n': return None
        body+=data[pos:pos+size]
        if len(body)>x['limit']: return None
        pos+=size+2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('hexadecimal-size fixture 0', json.loads(json.dumps(solve({'wire':'A;foo=bar\r\n0123456789\r\n0\r\nX:y\r\n\r\n','start':2,'stop':6,'limit':20}))), json.loads(json.dumps(['2345',10,['X:y']])))
check('hexadecimal-size fixture 1', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n3\r\ncde\r\n0\r\n\r\n','start':1,'stop':4,'limit':5}))), json.loads(json.dumps(['bcd',5,[]])))
check('hexadecimal-size fixture 2', json.loads(json.dumps(solve({'wire':'0\r\n\r\n','start':0,'stop':3,'limit':0}))), json.loads(json.dumps(['',0,[]])))
check('hexadecimal-size fixture 3', json.loads(json.dumps(solve({'wire':'1;z=q\r\nx\r\n0\r\n\r\n','start':0,'stop':1,'limit':1}))), json.loads(json.dumps(['x',1,[]])))
check('hexadecimal-size fixture 4', json.loads(json.dumps(solve({'wire':'1\r\nxXX0\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 5', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\n\r\nextra','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 6', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nBadTrailer\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 7', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n2\r\ncd\r\n0\r\n\r\n','start':0,'stop':4,'limit':3}))), json.loads(json.dumps(None)))
check('hexadecimal-size fixture 8', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nX:a\r\nY:b\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(['x',1,['X:a','Y:b']])))
check('hexadecimal-size fixture 9', json.loads(json.dumps(solve({'wire':format(N,"x")+"\r\n"+"p"*N+"\r\n0\r\n\r\n",'start':0,'stop':N,'limit':10}))), json.loads(json.dumps(["p"*N,N,[]])))
check('hexadecimal-size fixture 10', json.loads(json.dumps(solve({'wire':'10\r\nabcdefghijklmnop\r\n0\r\n\r\n','start':12,'stop':16,'limit':16}))), json.loads(json.dumps(['mnop',16,[]])))
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 range client removes transfer framing before applying representation offsets: chunk lengths use hexadecimal representation.

Decode a complete bounded ASCII chunked origin body. Size lines use 1..2 hexadecimal digits with optional semicolon extensions, followed by exact CRLF; each data chunk has exact trailing CRLF. A zero chunk ends data and introduces colon-bearing trailer lines terminated by an empty CRLF line. No bytes may follow trailers. Enforce cumulative decoded-byte limit x.limit. Return [decoded[start:stop],decoded-length,trailer-lines] or None. This controlled framing model does not implement general HTTP headers.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['wire']
    pos=0
    body=''
    trailers=[]
    while True:
        end=data.find('\r\n',pos)
        if end<0: return None
        digits=data[pos:end]
        if not 1<=len(digits)<=2 or any(c not in '0123456789abcdefABCDEF' for c in digits): return None
        size=int(digits,16)
        pos=end+2
        if size==0:
            while True:
                end=data.find('\r\n',pos)
                if end<0: return None
                line=data[pos:end]
                pos=end+2
                if not line: break
                if ':' not in line: return None
                trailers.append(line)
            if pos!=len(data): return None
            return [body[x['start']:x['stop']],len(body),trailers]
        if pos+size+2>len(data): return None
        if data[pos+size:pos+size+2]!='\r\n': return None
        body+=data[pos:pos+size]
        if len(body)>x['limit']: return None
        pos+=size+2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('size-extension fixture 0', json.loads(json.dumps(solve({'wire':'A;foo=bar\r\n0123456789\r\n0\r\nX:y\r\n\r\n','start':2,'stop':6,'limit':20}))), json.loads(json.dumps(['2345',10,['X:y']])))
check('size-extension fixture 1', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n3\r\ncde\r\n0\r\n\r\n','start':1,'stop':4,'limit':5}))), json.loads(json.dumps(['bcd',5,[]])))
check('size-extension fixture 2', json.loads(json.dumps(solve({'wire':'0\r\n\r\n','start':0,'stop':3,'limit':0}))), json.loads(json.dumps(['',0,[]])))
check('size-extension fixture 3', json.loads(json.dumps(solve({'wire':'1;z=q\r\nx\r\n0\r\n\r\n','start':0,'stop':1,'limit':1}))), json.loads(json.dumps(['x',1,[]])))
check('size-extension fixture 4', json.loads(json.dumps(solve({'wire':'1\r\nxXX0\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 5', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\n\r\nextra','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 6', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nBadTrailer\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 7', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n2\r\ncd\r\n0\r\n\r\n','start':0,'stop':4,'limit':3}))), json.loads(json.dumps(None)))
check('size-extension fixture 8', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nX:a\r\nY:b\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(['x',1,['X:a','Y:b']])))
check('size-extension fixture 9', json.loads(json.dumps(solve({'wire':format(N,"x")+"\r\n"+"p"*N+"\r\n0\r\n\r\n",'start':0,'stop':N,'limit':10}))), json.loads(json.dumps(["p"*N,N,[]])))
check('size-extension fixture 10', json.loads(json.dumps(solve({'wire':'10\r\nabcdefghijklmnop\r\n0\r\n\r\n','start':12,'stop':16,'limit':16}))), json.loads(json.dumps(['mnop',16,[]])))
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(x):
    data=x['wire']
    pos=0
    body=''
    trailers=[]
    while True:
        end=data.find('\r\n',pos)
        if end<0: return None
        digits=data[pos:end].split(';foo=',1)[0]
        if not 1<=len(digits)<=2 or any(c not in '0123456789abcdefABCDEF' for c in digits): return None
        size=int(digits,16)
        pos=end+2
        if size==0:
            while True:
                end=data.find('\r\n',pos)
                if end<0: return None
                line=data[pos:end]
                pos=end+2
                if not line: break
                if ':' not in line: return None
                trailers.append(line)
            if pos!=len(data): return None
            return [body[x['start']:x['stop']],len(body),trailers]
        if pos+size+2>len(data): return None
        if data[pos+size:pos+size+2]!='\r\n': return None
        body+=data[pos:pos+size]
        if len(body)>x['limit']: return None
        pos+=size+2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('size-extension fixture 0', json.loads(json.dumps(solve({'wire':'A;foo=bar\r\n0123456789\r\n0\r\nX:y\r\n\r\n','start':2,'stop':6,'limit':20}))), json.loads(json.dumps(['2345',10,['X:y']])))
check('size-extension fixture 1', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n3\r\ncde\r\n0\r\n\r\n','start':1,'stop':4,'limit':5}))), json.loads(json.dumps(['bcd',5,[]])))
check('size-extension fixture 2', json.loads(json.dumps(solve({'wire':'0\r\n\r\n','start':0,'stop':3,'limit':0}))), json.loads(json.dumps(['',0,[]])))
check('size-extension fixture 3', json.loads(json.dumps(solve({'wire':'1;z=q\r\nx\r\n0\r\n\r\n','start':0,'stop':1,'limit':1}))), json.loads(json.dumps(['x',1,[]])))
check('size-extension fixture 4', json.loads(json.dumps(solve({'wire':'1\r\nxXX0\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 5', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\n\r\nextra','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 6', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nBadTrailer\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(None)))
check('size-extension fixture 7', json.loads(json.dumps(solve({'wire':'2\r\nab\r\n2\r\ncd\r\n0\r\n\r\n','start':0,'stop':4,'limit':3}))), json.loads(json.dumps(None)))
check('size-extension fixture 8', json.loads(json.dumps(solve({'wire':'1\r\nx\r\n0\r\nX:a\r\nY:b\r\n\r\n','start':0,'stop':1,'limit':2}))), json.loads(json.dumps(['x',1,['X:a','Y:b']])))
check('size-extension fixture 9', json.loads(json.dumps(solve({'wire':format(N,"x")+"\r\n"+"p"*N+"\r\n0\r\n\r\n",'start':0,'stop':N,'limit':10}))), json.loads(json.dumps(["p"*N,N,[]])))
check('size-extension fixture 10', json.loads(json.dumps(solve({'wire':'10\r\nabcdefghijklmnop\r\n0\r\n\r\n','start':12,'stop':16,'limit':16}))), json.loads(json.dumps(['mnop',16,[]])))
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 range client removes transfer framing before applying representation offsets: chunk extensions do not become part of the size token.

Decode a complete bounded ASCII chunked origin body. Size lines use 1..2 hexadecimal digits with optional semicolon extensions, followed by exact CRLF; each data chunk has exact trailing CRLF. A zero chunk ends data and introduces colon-bearing trailer lines terminated by an empty CRLF line. No bytes may follow trailers. Enforce cumulative decoded-byte limit x.limit. Return [decoded[start:stop],decoded-length,trailer-lines] or None. This controlled framing model does not implement general HTTP headers.

split: open-access