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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else 1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('terminator-width fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('terminator-width fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('terminator-width fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('terminator-width fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('terminator-width fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('terminator-width fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('terminator-width fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('terminator-width fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('terminator-width fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('terminator-width fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('terminator-width fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('terminator-width fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('terminator-width fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: CRLF consumes two physical representation bytes.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if True:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('blank-records fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('blank-records fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('blank-records fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('blank-records fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('blank-records fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('blank-records fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('blank-records fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('blank-records fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('blank-records fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body or terminator==2:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('blank-records fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('blank-records fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('blank-records fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('blank-records fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('blank-records fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('blank-records fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('blank-records fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('blank-records fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('blank-records fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: empty line separators do not allocate logical record numbers.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body.decode('utf-8'))>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('record-byte-cap fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('record-byte-cap fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('record-byte-cap fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-cap fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('record-byte-cap fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-cap fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']*2: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('record-byte-cap fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('record-byte-cap fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('record-byte-cap fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-cap fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('record-byte-cap fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('record-byte-cap fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-cap fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: multibyte record contents count toward the byte-index cap.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body.decode('utf-8')),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('record-byte-length fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('record-byte-length fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('record-byte-length fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-length fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('record-byte-length fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-length fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),1])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('record-byte-length fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('record-byte-length fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('record-byte-length fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-length fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('record-byte-length fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('record-byte-length fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('record-byte-length fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: index entries expose octet counts rather than codepoint counts.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+2
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('next-line-position fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('next-line-position fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('next-line-position fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('next-line-position fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('next-line-position fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('next-line-position fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+terminator
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('next-line-position fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('next-line-position fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('next-line-position fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('next-line-position fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('next-line-position fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('next-line-position fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('next-line-position fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: physical row scanning advances past the complete line terminator.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('row-index-bounds fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('row-index-bounds fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('row-index-bounds fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('row-index-bounds fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('row-index-bounds fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('row-index-bounds fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<-1 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in x['select']]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('row-index-bounds fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('row-index-bounds fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('row-index-bounds fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('row-index-bounds fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('row-index-bounds fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('row-index-bounds fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('row-index-bounds fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: negative row indices are invalid rather than Python tail indexing.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in sorted(set(x['select']))]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('selection-multiplicity fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('selection-multiplicity fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('selection-multiplicity fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('selection-multiplicity fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('selection-multiplicity fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('selection-multiplicity fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(x):
    data=x['text'].encode('utf-8')
    pos=3 if data.startswith(b'\xef\xbb\xbf') else 0
    entries=[]
    bodies=[]
    while pos<len(data):
        end=data.find(b'\n',pos)
        final=end<0
        end=len(data) if final else end
        content_end=end-1 if not final and end>pos and data[end-1]==13 else end
        terminator=0 if final else end-content_end+1
        body=data[pos:content_end]
        if body:
            if len(body)>x['cap']: return None
            entries.append([pos,len(body),terminator])
            bodies.append(body.hex())
        pos=len(data) if final else end+1
    if any(i<0 or i>=len(bodies) for i in x['select']): return None
    return [entries,[bodies[i] for i in dict.fromkeys(x['select'])]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('selection-multiplicity fixture 0', json.loads(json.dumps(solve({'text':'a\r\nb\n\ncd','cap':5,'select':[2,0,2]}))), json.loads(json.dumps([[[0,1,2],[3,1,1],[6,2,0]],['6364','61','6364']])))
check('selection-multiplicity fixture 1', json.loads(json.dumps(solve({'text':'\ufeffa\n','cap':1,'select':[0]}))), json.loads(json.dumps([[[3,1,1]],['61']])))
check('selection-multiplicity fixture 2', json.loads(json.dumps(solve({'text':'\n\r\n','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('selection-multiplicity fixture 3', json.loads(json.dumps(solve({'text':'abc','cap':2,'select':[]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 4', json.loads(json.dumps(solve({'text':'é','cap':1,'select':[]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 5', json.loads(json.dumps(solve({'text':"é"*N+"\n",'cap':2*N,'select':[0]}))), json.loads(json.dumps([[[0,2*N,1]],["c3a9"*N]])))
check('selection-multiplicity fixture 6', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[-1]}))), json.loads(json.dumps(None)))
check('selection-multiplicity fixture 7', json.loads(json.dumps(solve({'text':'','cap':0,'select':[]}))), json.loads(json.dumps([[],[]])))
check('selection-multiplicity fixture 8', json.loads(json.dumps(solve({'text':'a\n','cap':1,'select':[1]}))), json.loads(json.dumps(None)))
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A row-range endpoint builds byte-addressable line record metadata: repeat row selectors retain repeated response bodies.

Build a bounded row index over UTF-8 text. A leading UTF-8 BOM is excluded from records but occupies3 representation bytes. LF or CRLF terminates a row; blank rows are ignored. A final unterminated nonempty row is included. Return [entries,selected-hex-bodies], with each entry [absolute-byte-start,content-byte-count,terminator-byte-count]. Enforce a per-record byte cap and nonnegative selected row indices. Selection order and multiplicity are meaningful.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if False:
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('terminal-event fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('terminal-event fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('terminal-event fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('terminal-event fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('terminal-event fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('terminal-event fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('terminal-event fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('terminal-event fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state=='aborted':
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('terminal-event fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('terminal-event fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('terminal-event fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('terminal-event fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('terminal-event fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('terminal-event fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('terminal-event fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('terminal-event fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: late syscall completions do not reopen completed range streams.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('negative-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('negative-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('negative-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('negative-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('negative-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('negative-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('negative-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('negative-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<-1 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('negative-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('negative-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('negative-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('negative-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('negative-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('negative-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('negative-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('negative-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: invalid negative syscall counts cannot rewind a range cursor.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('oversized-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('oversized-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('oversized-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('oversized-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('oversized-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('oversized-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('oversized-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('oversized-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining+1:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('oversized-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('oversized-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('oversized-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('oversized-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('oversized-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('oversized-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('oversized-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('oversized-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: kernel reports beyond the promised range abort before state advances.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted if error!='interrupt' else 0
        remaining-=accepted if error!='interrupt' else 0
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('interrupt-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('interrupt-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('interrupt-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('interrupt-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('interrupt-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('interrupt-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted if error=='none' else 0
        remaining-=accepted if error=='none' else 0
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('interrupt-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('interrupt-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('interrupt-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('interrupt-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('interrupt-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('interrupt-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: successful byte progress survives an interrupted syscall status.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted if error!='again' else 0
        remaining-=accepted if error!='again' else 0
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('again-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('again-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('again-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('again-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('again-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('again-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('again-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('again-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted if error!='again' else 0
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('again-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('again-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('again-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('again-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('again-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('again-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('again-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('again-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: would-block may accompany successfully transmitted range bytes.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='running'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('fatal-after-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('fatal-after-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('fatal-after-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('fatal-after-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('fatal-after-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('fatal-after-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('fatal-after-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('fatal-after-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='complete' if remaining==0 else 'aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('fatal-after-progress fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('fatal-after-progress fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('fatal-after-progress fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('fatal-after-progress fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('fatal-after-progress fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('fatal-after-progress fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('fatal-after-progress fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('fatal-after-progress fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: fatal errors remain terminal even after their last accepted bytes.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0 and error=='none': state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('completed-would-block fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('completed-would-block fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('completed-would-block fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('completed-would-block fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('completed-would-block fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('completed-would-block fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('completed-would-block fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('completed-would-block fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0 and error!='again': state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('completed-would-block fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('completed-would-block fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('completed-would-block fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('completed-would-block fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('completed-would-block fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('completed-would-block fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('completed-would-block fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('completed-would-block fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: all promised bytes complete even if the same syscall reports would-block.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='truncated'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('interrupt-zero fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('interrupt-zero fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('interrupt-zero fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('interrupt-zero fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-zero fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-zero fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('interrupt-zero fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('interrupt-zero fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='blocked'
        elif accepted==0: state='truncated'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('interrupt-zero fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('interrupt-zero fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('interrupt-zero fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('interrupt-zero fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-zero fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('interrupt-zero fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('interrupt-zero fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('interrupt-zero fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: an interrupted zero-byte syscall is rescheduled without declaring EOF.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access


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

N = 1
observations = []
def solve(x):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='running'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('zero-success-eof fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('zero-success-eof fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('zero-success-eof fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('zero-success-eof fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('zero-success-eof fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('zero-success-eof fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('zero-success-eof fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('zero-success-eof fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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):
    offset=x['offset']
    remaining=x['remaining']
    state='running'
    opened=True
    out=[]
    for accepted,error in x['events']:
        if state in ('complete','aborted','truncated'):
            out.append([offset,remaining,state,opened])
            continue
        if accepted<0 or accepted>remaining:
            state='aborted'
            opened=False
            out.append([offset,remaining,state,opened])
            continue
        offset+=accepted
        remaining-=accepted
        if error=='fatal': state='aborted'
        elif remaining==0: state='complete'
        elif error=='again': state='blocked'
        elif error=='interrupt': state='running'
        elif accepted==0: state='blocked'
        else: state='running'
        opened=state not in ('complete','aborted','truncated')
        out.append([offset,remaining,state,opened])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('zero-success-eof fixture 0', json.loads(json.dumps(solve({'offset':10,'remaining':5,'events':[[2,'interrupt'],[1,'again'],[2,'none']]}))), json.loads(json.dumps([[12,3,'running',True],[13,2,'blocked',True],[15,0,'complete',False]])))
check('zero-success-eof fixture 1', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[0,'again'],[0,'interrupt'],[0,'none'],[1,'none']]}))), json.loads(json.dumps([[0,2,'blocked',True],[0,2,'running',True],[0,2,'truncated',False],[0,2,'truncated',False]])))
check('zero-success-eof fixture 2', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[1,'fatal'],[1,'none']]}))), json.loads(json.dumps([[1,1,'aborted',False],[1,1,'aborted',False]])))
check('zero-success-eof fixture 3', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[-1,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('zero-success-eof fixture 4', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[3,'none']]}))), json.loads(json.dumps([[0,2,'aborted',False]])))
check('zero-success-eof fixture 5', json.loads(json.dumps(solve({'offset':N,'remaining':N,'events':[[N,'again'],[0,'fatal']]}))), json.loads(json.dumps([[2*N,0,'complete',False],[2*N,0,'complete',False]])))
check('zero-success-eof fixture 6', json.loads(json.dumps(solve({'offset':0,'remaining':0,'events':[[0,'none']]}))), json.loads(json.dumps([[0,0,'complete',False]])))
check('zero-success-eof fixture 7', json.loads(json.dumps(solve({'offset':0,'remaining':2,'events':[[2,'fatal']]}))), json.loads(json.dumps([[2,0,'aborted',False]])))
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 zero-copy range sender accounts for partial progress reported with syscall errors: a clean zero-byte return with remaining bytes is truncation.

x contains absolute offset, remaining promised bytes, and syscall events [accepted,error], where error is none,again,interrupt,fatal. Counts outside0..remaining abort without advancing. Valid accepted bytes advance even when again or interrupt accompanies them. Fatal errors abort; zero remaining completes; again blocks; interrupt reschedules; zero bytes with none truncates. Terminal states ignore later events. Return state snapshots [offset,remaining,state,handle-open] after each event.

split: open-access