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

N = 1
observations = []
def solve(x):
    cells=dict(x['cells'])
    r=x['request']
    if r['method']!='PUT': return ['method',sorted(cells.items())]
    if r['session']!=x['session']: return ['session',sorted(cells.items())]
    if r['generation']!=x['generation']: return ['generation',sorted(cells.items())]
    if r['total']!=x['total']: return ['total',sorted(cells.items())]
    if r['total']>x['quota']: return ['quota',sorted(cells.items())]
    if r['start']<0 or r['stop']<r['start'] or r['stop']>r['total']: return ['bounds',sorted(cells.items())]
    if len(r['body'])!=r['stop']-r['start']: return ['length',sorted(cells.items())]
    if any(i in cells and cells[i]!=c for i,c in enumerate(r['body'],r['start'])): return ['conflict',sorted(cells.items())]
    for i,c in enumerate(r['body']): cells[i]=c
    complete=set(cells)==set(range(x['total']))
    return ['complete' if complete else 'partial',sorted(cells.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('write-position fixture 0', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'bc'}}))), json.loads(json.dumps(['complete',[[0,'a'],[1,'b'],[2,'c']]])))
check('write-position fixture 1', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'POST','session':'s','generation':2,'total':3,'start':1,'stop':2,'body':'b'}}))), json.loads(json.dumps(['method',[[0,'a']]])))
check('write-position fixture 2', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'other','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['session',[]])))
check('write-position fixture 3', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':3,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['generation',[]])))
check('write-position fixture 4', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':4,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['total',[]])))
check('write-position fixture 5', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':2,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['quota',[]])))
check('write-position fixture 6', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':-1,'stop':0,'body':'a'}}))), json.loads(json.dumps(['bounds',[]])))
check('write-position fixture 7', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'b'}}))), json.loads(json.dumps(['length',[]])))
check('write-position fixture 8', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'x'}}))), json.loads(json.dumps(['conflict',[[0,'a']]])))
check('write-position fixture 9', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['partial',[[0,'a']]])))
check('write-position fixture 10', json.loads(json.dumps(solve({'session':'s','generation':2,'total':N,'quota':10,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':N,'start':0,'stop':N,'body':"p"*N}}))), json.loads(json.dumps(['complete',[[i,"p"] for i in range(N)]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    cells=dict(x['cells'])
    r=x['request']
    if r['method']!='PUT': return ['method',sorted(cells.items())]
    if r['session']!=x['session']: return ['session',sorted(cells.items())]
    if r['generation']!=x['generation']: return ['generation',sorted(cells.items())]
    if r['total']!=x['total']: return ['total',sorted(cells.items())]
    if r['total']>x['quota']: return ['quota',sorted(cells.items())]
    if r['start']<0 or r['stop']<r['start'] or r['stop']>r['total']: return ['bounds',sorted(cells.items())]
    if len(r['body'])!=r['stop']-r['start']: return ['length',sorted(cells.items())]
    if any(i in cells and cells[i]!=c for i,c in enumerate(r['body'],r['start'])): return ['conflict',sorted(cells.items())]
    for i,c in enumerate(r['body'],len(cells)): cells[i]=c
    complete=set(cells)==set(range(x['total']))
    return ['complete' if complete else 'partial',sorted(cells.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('write-position fixture 0', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'bc'}}))), json.loads(json.dumps(['complete',[[0,'a'],[1,'b'],[2,'c']]])))
check('write-position fixture 1', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'POST','session':'s','generation':2,'total':3,'start':1,'stop':2,'body':'b'}}))), json.loads(json.dumps(['method',[[0,'a']]])))
check('write-position fixture 2', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'other','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['session',[]])))
check('write-position fixture 3', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':3,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['generation',[]])))
check('write-position fixture 4', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':4,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['total',[]])))
check('write-position fixture 5', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':2,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['quota',[]])))
check('write-position fixture 6', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':-1,'stop':0,'body':'a'}}))), json.loads(json.dumps(['bounds',[]])))
check('write-position fixture 7', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'b'}}))), json.loads(json.dumps(['length',[]])))
check('write-position fixture 8', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'x'}}))), json.loads(json.dumps(['conflict',[[0,'a']]])))
check('write-position fixture 9', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['partial',[[0,'a']]])))
check('write-position fixture 10', json.loads(json.dumps(solve({'session':'s','generation':2,'total':N,'quota':10,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':N,'start':0,'stop':N,'body':"p"*N}}))), json.loads(json.dumps(['complete',[[i,"p"] for i in range(N)]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A bounded Content-Range upload endpoint validates chunk admission: admitted upload chunks write at declared absolute offsets.

This stipulated endpoint accepts PUT chunks against an upload session. Inputs include session ID, generation, declared complete length, [start,stop) byte span, ASCII body, and known committed cells. Return a disposition and updated cells; malformed metadata never mutates stored bytes. Exact duplicate chunks are idempotent, contradicting overlap is conflict, and quota concerns the complete representation rather than chunk size.

split: open-access


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

N = 1
observations = []
def solve(x):
    cells=dict(x['cells'])
    r=x['request']
    if r['method']!='PUT': return ['method',sorted(cells.items())]
    if r['session']!=x['session']: return ['session',sorted(cells.items())]
    if r['generation']!=x['generation']: return ['generation',sorted(cells.items())]
    if r['total']!=x['total']: return ['total',sorted(cells.items())]
    if r['total']>x['quota']: return ['quota',sorted(cells.items())]
    if r['start']<0 or r['stop']<r['start'] or r['stop']>r['total']: return ['bounds',sorted(cells.items())]
    if len(r['body'])!=r['stop']-r['start']: return ['length',sorted(cells.items())]
    if any(i in cells and cells[i]!=c for i,c in enumerate(r['body'],r['start'])): return ['conflict',sorted(cells.items())]
    for i,c in enumerate(r['body'],r['start']): cells[i]=c
    complete=bool(cells)
    return ['complete' if complete else 'partial',sorted(cells.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('completion-coverage fixture 0', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'bc'}}))), json.loads(json.dumps(['complete',[[0,'a'],[1,'b'],[2,'c']]])))
check('completion-coverage fixture 1', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'POST','session':'s','generation':2,'total':3,'start':1,'stop':2,'body':'b'}}))), json.loads(json.dumps(['method',[[0,'a']]])))
check('completion-coverage fixture 2', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'other','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['session',[]])))
check('completion-coverage fixture 3', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':3,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['generation',[]])))
check('completion-coverage fixture 4', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':4,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['total',[]])))
check('completion-coverage fixture 5', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':2,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['quota',[]])))
check('completion-coverage fixture 6', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':-1,'stop':0,'body':'a'}}))), json.loads(json.dumps(['bounds',[]])))
check('completion-coverage fixture 7', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'b'}}))), json.loads(json.dumps(['length',[]])))
check('completion-coverage fixture 8', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'x'}}))), json.loads(json.dumps(['conflict',[[0,'a']]])))
check('completion-coverage fixture 9', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['partial',[[0,'a']]])))
check('completion-coverage fixture 10', json.loads(json.dumps(solve({'session':'s','generation':2,'total':N,'quota':10,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':N,'start':0,'stop':N,'body':"p"*N}}))), json.loads(json.dumps(['complete',[[i,"p"] for i in range(N)]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

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

N = 1
observations = []
def solve(x):
    cells=dict(x['cells'])
    r=x['request']
    if r['method']!='PUT': return ['method',sorted(cells.items())]
    if r['session']!=x['session']: return ['session',sorted(cells.items())]
    if r['generation']!=x['generation']: return ['generation',sorted(cells.items())]
    if r['total']!=x['total']: return ['total',sorted(cells.items())]
    if r['total']>x['quota']: return ['quota',sorted(cells.items())]
    if r['start']<0 or r['stop']<r['start'] or r['stop']>r['total']: return ['bounds',sorted(cells.items())]
    if len(r['body'])!=r['stop']-r['start']: return ['length',sorted(cells.items())]
    if any(i in cells and cells[i]!=c for i,c in enumerate(r['body'],r['start'])): return ['conflict',sorted(cells.items())]
    for i,c in enumerate(r['body'],r['start']): cells[i]=c
    complete=len(r['body'])==r['stop']-r['start']
    return ['complete' if complete else 'partial',sorted(cells.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('completion-coverage fixture 0', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'bc'}}))), json.loads(json.dumps(['complete',[[0,'a'],[1,'b'],[2,'c']]])))
check('completion-coverage fixture 1', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'POST','session':'s','generation':2,'total':3,'start':1,'stop':2,'body':'b'}}))), json.loads(json.dumps(['method',[[0,'a']]])))
check('completion-coverage fixture 2', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'other','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['session',[]])))
check('completion-coverage fixture 3', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':3,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['generation',[]])))
check('completion-coverage fixture 4', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':4,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['total',[]])))
check('completion-coverage fixture 5', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':2,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['quota',[]])))
check('completion-coverage fixture 6', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':-1,'stop':0,'body':'a'}}))), json.loads(json.dumps(['bounds',[]])))
check('completion-coverage fixture 7', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':1,'stop':3,'body':'b'}}))), json.loads(json.dumps(['length',[]])))
check('completion-coverage fixture 8', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'x'}}))), json.loads(json.dumps(['conflict',[[0,'a']]])))
check('completion-coverage fixture 9', json.loads(json.dumps(solve({'session':'s','generation':2,'total':3,'quota':5,'cells':[[0,'a']],'request':{'method':'PUT','session':'s','generation':2,'total':3,'start':0,'stop':1,'body':'a'}}))), json.loads(json.dumps(['partial',[[0,'a']]])))
check('completion-coverage fixture 10', json.loads(json.dumps(solve({'session':'s','generation':2,'total':N,'quota':10,'cells':[],'request':{'method':'PUT','session':'s','generation':2,'total':N,'start':0,'stop':N,'body':"p"*N}}))), json.loads(json.dumps(['complete',[[i,"p"] for i in range(N)]])))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A bounded Content-Range upload endpoint validates chunk admission: a nonempty accepted chunk is not necessarily a complete upload.

This stipulated endpoint accepts PUT chunks against an upload session. Inputs include session ID, generation, declared complete length, [start,stop) byte span, ASCII body, and known committed cells. Return a disposition and updated cells; malformed metadata never mutates stored bytes. Exact duplicate chunks are idempotent, contradicting overlap is conflict, and quota concerns the complete representation rather than chunk size.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=2
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('segment-advance fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('segment-advance fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('segment-advance fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('segment-advance fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('segment-advance fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('segment-advance fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('segment-advance fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1 if segment else 2
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('segment-advance fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('segment-advance fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('segment-advance fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('segment-advance fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('segment-advance fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('segment-advance fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('segment-advance fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: finishing a segment advances exactly one framing component.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=min(offset,len(segments[index])) if index<len(segments) else 0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offset-reset fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('offset-reset fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('offset-reset fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('offset-reset fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('offset-reset fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('offset-reset fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('offset-reset fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0 if index==len(segments) else min(1,len(segments[index]))
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offset-reset fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('offset-reset fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('offset-reset fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('offset-reset fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('offset-reset fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('offset-reset fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('offset-reset fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: a new framing component starts at its own zero offset.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: budget=1
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('zero-credit fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('zero-credit fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('zero-credit fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('zero-credit fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('zero-credit fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('zero-credit fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('zero-credit fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0 and sent: break
            if budget==0: budget=1
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('zero-credit fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('zero-credit fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('zero-credit fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('zero-credit fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('zero-credit fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('zero-credit fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('zero-credit fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: zero transport credit cannot emit a pending delimiter byte.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment))
            amount=min(amount,len(segment)-offset) if offset==0 else min(amount,len(segment)-offset+1)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-segment fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('remaining-segment fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('remaining-segment fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('remaining-segment fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('remaining-segment fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('remaining-segment fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('remaining-segment fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,max(0,len(segment)-offset-1))
            if amount==0: break
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('remaining-segment fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('remaining-segment fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('remaining-segment fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('remaining-segment fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('remaining-segment fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('remaining-segment fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('remaining-segment fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: drain size considers the remaining suffix instead of whole segment length.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[:amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('slice-origin fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('slice-origin fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('slice-origin fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('slice-origin fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('slice-origin fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('slice-origin fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('slice-origin fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[-amount:]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('slice-origin fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('slice-origin fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('slice-origin fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('slice-origin fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('slice-origin fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('slice-origin fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('slice-origin fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: a partially sent header resumes at its saved intra-header position.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset=amount
            if amount==0: break
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offset-progress fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('offset-progress fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('offset-progress fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('offset-progress fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('offset-progress fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('offset-progress fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('offset-progress fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount if offset==0 else amount+1
            offset=min(offset,len(segment))
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offset-progress fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('offset-progress fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('offset-progress fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('offset-progress fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('offset-progress fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('offset-progress fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('offset-progress fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: accepted bytes advance the cursor rather than replacing it.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget=0
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('credit-progress fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('credit-progress fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('credit-progress fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('credit-progress fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('credit-progress fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('credit-progress fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('credit-progress fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget=max(0,budget-len(segment))
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('credit-progress fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('credit-progress fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('credit-progress fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('credit-progress fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('credit-progress fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('credit-progress fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('credit-progress fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: transport credit is decremented by actual accepted bytes.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('window-output fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('window-output fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('window-output fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('window-output fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('window-output fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('window-output fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('window-output fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent=(sent if offset else '')+segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('window-output fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('window-output fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('window-output fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('window-output fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('window-output fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('window-output fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('window-output fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: each transport window retains bytes emitted from all reached components.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment) and budget>0:
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('empty-segments fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('empty-segments fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('empty-segments fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('empty-segments fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('empty-segments fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('empty-segments fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('empty-segments fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment) and (segment or budget>0):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('empty-segments fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('empty-segments fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('empty-segments fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('empty-segments fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('empty-segments fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('empty-segments fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('empty-segments fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: zero-length framing components do not consume transport credit.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

split: open-access


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

N = 1
observations = []
def solve(x):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        if sent: out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('window-retention fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('window-retention fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('window-retention fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('window-retention fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('window-retention fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('window-retention fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('window-retention fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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):
    segments=x['segments']
    index=0
    offset=0
    out=[]
    for budget in x['budgets']:
        sent=''
        while index<len(segments):
            segment=segments[index]
            if offset>=len(segment):
                index+=1
                offset=0
                continue
            if budget==0: break
            amount=min(budget,len(segment)-offset)
            sent+=segment[offset:offset+amount]
            offset+=amount
            budget-=amount
        if sent or budget>0: out.append(sent)
    return [out,index,offset]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('window-retention fixture 0', json.loads(json.dumps(solve({'segments':['--b','H','BODY','\r\n'],'budgets':[2,2,3,3]}))), json.loads(json.dumps([['--','bH','BOD','Y\r\n'],4,0])))
check('window-retention fixture 1', json.loads(json.dumps(solve({'segments':['A','BC'],'budgets':[1,0,1,1,1]}))), json.loads(json.dumps([['A','','B','C',''],2,0])))
check('window-retention fixture 2', json.loads(json.dumps(solve({'segments':['','A','','BC',''],'budgets':[0,2,1]}))), json.loads(json.dumps([['','AB','C'],5,0])))
check('window-retention fixture 3', json.loads(json.dumps(solve({'segments':[],'budgets':[0,1]}))), json.loads(json.dumps([['',''],0,0])))
check('window-retention fixture 4', json.loads(json.dumps(solve({'segments':['ABCDE'],'budgets':[2,1]}))), json.loads(json.dumps([['AB','C'],0,3])))
check('window-retention fixture 5', json.loads(json.dumps(solve({'segments':["p"*N,'X'],'budgets':[N,1]}))), json.loads(json.dumps([["p"*N,'X'],2,0])))
check('window-retention fixture 6', json.loads(json.dumps(solve({'segments':['AB','CDE'],'budgets':[4]}))), json.loads(json.dumps([['ABCD'],1,2])))
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 multipart sender resumes framing and payload segments across transport credit windows: zero-byte windows remain visible to the transport caller.

x has nonempty ASCII segments representing alternating boundary/header/body/framing bytes, and nonnegative write budgets. Drain in order, consuming each budget across segment boundaries. Return [one emitted string per budget,segment-index,intra-segment-offset]. Empty segments are skipped even with zero budget. A exhausted plan is stable. These preframed segments isolate cursor management from wire-format construction.

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 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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cancelled-job 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('cancelled-job 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('cancelled-job 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('cancelled-job 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('cancelled-job fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('cancelled-job 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('cancelled-job 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('cancelled-job 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 and 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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cancelled-job 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('cancelled-job 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('cancelled-job 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('cancelled-job 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('cancelled-job fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('cancelled-job 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('cancelled-job 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('cancelled-job 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: canceled consumers do not schedule more origin byte reads.

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
        order.sort()
        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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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('client-appearance-order fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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()
        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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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('client-appearance-order fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('client-appearance-order 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('client-appearance-order 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('client-appearance-order 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: dispatch order follows consumer arrival rather than client names.

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].insert(0,[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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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('per-client-fifo fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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]=[[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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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('per-client-fifo fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('per-client-fifo 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('per-client-fifo 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('per-client-fifo 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: later ranges cannot displace an earlier request from the same consumer.

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']+1):
        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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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('dispatch-slots fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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(max(1,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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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('dispatch-slots fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('dispatch-slots 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('dispatch-slots 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('dispatch-slots 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: a scheduling turn does not overrun its origin request slot allowance.

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=count
        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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('chunk-size 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('chunk-size 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('chunk-size 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('chunk-size 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('chunk-size fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('chunk-size 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('chunk-size 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('chunk-size 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=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]]
    return [out,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('chunk-size 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('chunk-size 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('chunk-size 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('chunk-size 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('chunk-size fixture 4', json.loads(json.dumps(solve({'jobs':[['a',0,2,False]],'chunk':2,'slots':0}))), json.loads(json.dumps([[],[['a',0,2]]])))
check('chunk-size 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('chunk-size 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('chunk-size 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: large byte requests are bounded by the origin chunk policy.

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