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

N = 1
observations = []
def solve(authorized, exists, part):
    return [404,""] if not exists else ([206,part] if authorized else [403,""])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('authorization-before-object-existence fixture 0', solve(False,False,"p"*N), [403,""])
check('authorization-before-object-existence fixture 1', solve(False,True,"p"), [403,""])
check('authorization-before-object-existence fixture 2', solve(True,False,"p"), [404,""])
check('authorization-before-object-existence fixture 3', solve(True,True,"p"*N), [206,"p"*N])
check('authorization-before-object-existence fixture 4', solve(False,False,""), [403,""])
check('authorization-before-object-existence fixture 5', solve(True,True,""), [206,""])
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(authorized, exists, part):
    return [404,""] if not authorized else ([206,part] if exists else [404,""])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('authorization-before-object-existence fixture 0', solve(False,False,"p"*N), [403,""])
check('authorization-before-object-existence fixture 1', solve(False,True,"p"), [403,""])
check('authorization-before-object-existence fixture 2', solve(True,False,"p"), [404,""])
check('authorization-before-object-existence fixture 3', solve(True,True,"p"*N), [206,"p"*N])
check('authorization-before-object-existence fixture 4', solve(False,False,""), [403,""])
check('authorization-before-object-existence fixture 5', solve(True,True,""), [206,""])
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:
Unauthorized range probes hide whether the object exists.

Local concealment policy returns [403,empty] for unauthorized users before object existence; authorized absent objects return 404, present objects 206.

split: open-access


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

N = 1
observations = []
def solve(candidate, parts):
    return candidate
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-boundary-collision fixture 0', solve("b"+str(N),["--b"+str(N),"--b"+str(N)+"x"]), "b"+str(N)+"xx")
check('multipart-boundary-collision fixture 1', solve("b",["abc--b"]), "b")
check('multipart-boundary-collision fixture 2', solve("b",["x\r\n--b"]), "bx")
check('multipart-boundary-collision fixture 3', solve("b",[]), "b")
check('multipart-boundary-collision fixture 4', solve("b",[""]), "b")
check('multipart-boundary-collision fixture 5', solve("b",["--other"]), "b")
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(candidate, parts):
    return candidate+"x" if any(("--"+candidate) in p for p in parts) else candidate
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-boundary-collision fixture 0', solve("b"+str(N),["--b"+str(N),"--b"+str(N)+"x"]), "b"+str(N)+"xx")
check('multipart-boundary-collision fixture 1', solve("b",["abc--b"]), "b")
check('multipart-boundary-collision fixture 2', solve("b",["x\r\n--b"]), "bx")
check('multipart-boundary-collision fixture 3', solve("b",[]), "b")
check('multipart-boundary-collision fixture 4', solve("b",[""]), "b")
check('multipart-boundary-collision fixture 5', solve("b",["--other"]), "b")
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:
Multipart delimiters cannot occur at payload line boundaries.

Choose candidate, candidate+x, ... until no part contains a line starting --candidate. Parts are ASCII strings; only start of body or after CRLF is a line boundary.

split: open-access


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

N = 1
observations = []
def solve(boundary, chunks):
    return "".join(chunks)+"--"+boundary+"\r\n"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-closing-delimiter fixture 0', solve("b"+str(N),[]), "--b"+str(N)+"--\r\n")
check('multipart-closing-delimiter fixture 1', solve("b",["a","c"]), "ac--b--\r\n")
check('multipart-closing-delimiter fixture 2', solve("b",[""]), "--b--\r\n")
check('multipart-closing-delimiter fixture 3', solve("x",["abc"]), "abc--x--\r\n")
check('multipart-closing-delimiter fixture 4', solve("z",["\r\n"]), "\r\n--z--\r\n")
check('multipart-closing-delimiter fixture 5', solve("b",["a"*N]), "a"*N+"--b--\r\n")
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(boundary, chunks):
    return "".join(chunks)+("--"+boundary+"--\r\n" if chunks else "")
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-closing-delimiter fixture 0', solve("b"+str(N),[]), "--b"+str(N)+"--\r\n")
check('multipart-closing-delimiter fixture 1', solve("b",["a","c"]), "ac--b--\r\n")
check('multipart-closing-delimiter fixture 2', solve("b",[""]), "--b--\r\n")
check('multipart-closing-delimiter fixture 3', solve("x",["abc"]), "abc--x--\r\n")
check('multipart-closing-delimiter fixture 4', solve("z",["\r\n"]), "\r\n--z--\r\n")
check('multipart-closing-delimiter fixture 5', solve("b",["a"*N]), "a"*N+"--b--\r\n")
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The multipart terminator closes the boundary exactly once.

Return serialized preframed chunks followed by --boundary-- CRLF, including an empty multipart body.

split: open-access


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

N = 1
observations = []
def solve(body):
    return body if body.endswith("\r\n") else body+"\r\n"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-payload-separator fixture 0', solve("x"*N+"\r\n"), "x"*N+"\r\n\r\n")
check('multipart-payload-separator fixture 1', solve(""), "\r\n")
check('multipart-payload-separator fixture 2', solve("x"*N), "x"*N+"\r\n")
check('multipart-payload-separator fixture 3', solve("\r\n\r\n"), "\r\n\r\n\r\n")
check('multipart-payload-separator fixture 4', solve("x\r"), "x\r\r\n")
check('multipart-payload-separator fixture 5', solve("x\n"), "x\n\r\n")
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(body):
    return body.rstrip("\r\n")+"\r\n"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-payload-separator fixture 0', solve("x"*N+"\r\n"), "x"*N+"\r\n\r\n")
check('multipart-payload-separator fixture 1', solve(""), "\r\n")
check('multipart-payload-separator fixture 2', solve("x"*N), "x"*N+"\r\n")
check('multipart-payload-separator fixture 3', solve("\r\n\r\n"), "\r\n\r\n\r\n")
check('multipart-payload-separator fixture 4', solve("x\r"), "x\r\r\n")
check('multipart-payload-separator fixture 5', solve("x\n"), "x\n\r\n")
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Part payloads always precede a framing CRLF.

Append one CRLF framing separator regardless of whether payload already ends in CRLF. The payload itself must remain intact.

split: open-access


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

N = 1
observations = []
def solve(boundary, object_type):
    return object_type
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-top-content-type fixture 0', solve("b"+str(N),"text/plain"), "multipart/byteranges; boundary=\"b"+str(N)+"\"")
check('multipart-top-content-type fixture 1', solve("x","image/png"), 'multipart/byteranges; boundary="x"')
check('multipart-top-content-type fixture 2', solve("","x/y"), 'multipart/byteranges; boundary=""')
check('multipart-top-content-type fixture 3', solve("abc",""), 'multipart/byteranges; boundary="abc"')
check('multipart-top-content-type fixture 4', solve("Z","application/octet-stream"), 'multipart/byteranges; boundary="Z"')
check('multipart-top-content-type fixture 5', solve("ab12","text/plain"), 'multipart/byteranges; boundary="ab12"')
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(boundary, object_type):
    return "multipart/byteranges"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-top-content-type fixture 0', solve("b"+str(N),"text/plain"), "multipart/byteranges; boundary=\"b"+str(N)+"\"")
check('multipart-top-content-type fixture 1', solve("x","image/png"), 'multipart/byteranges; boundary="x"')
check('multipart-top-content-type fixture 2', solve("","x/y"), 'multipart/byteranges; boundary=""')
check('multipart-top-content-type fixture 3', solve("abc",""), 'multipart/byteranges; boundary="abc"')
check('multipart-top-content-type fixture 4', solve("Z","application/octet-stream"), 'multipart/byteranges; boundary="Z"')
check('multipart-top-content-type fixture 5', solve("ab12","text/plain"), 'multipart/byteranges; boundary="ab12"')
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:
Multipart response type carries a boundary rather than the object media type.

Return the stipulated ASCII top-level type multipart/byteranges; boundary="value". Boundary has only letters/digits and may be empty in this framing-unit test.

split: open-access


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

N = 1
observations = []
def solve(object_type, count):
    return ["multipart/byteranges"]*count
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-part-type-preserved fixture 0', solve("text/plain; charset=utf-8",N), ["text/plain; charset=utf-8"]*N)
check('multipart-part-type-preserved fixture 1', solve("image/png",2), ["image/png","image/png"])
check('multipart-part-type-preserved fixture 2', solve("text/plain",0), [])
check('multipart-part-type-preserved fixture 3', solve("application/x; a=b",2), ["application/x; a=b"]*2)
check('multipart-part-type-preserved fixture 4', solve("x/y",1), ["x/y"])
check('multipart-part-type-preserved fixture 5', solve("",1), [""])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(object_type, count):
    return [object_type.split(";")[0]]*count
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-part-type-preserved fixture 0', solve("text/plain; charset=utf-8",N), ["text/plain; charset=utf-8"]*N)
check('multipart-part-type-preserved fixture 1', solve("image/png",2), ["image/png","image/png"])
check('multipart-part-type-preserved fixture 2', solve("text/plain",0), [])
check('multipart-part-type-preserved fixture 3', solve("application/x; a=b",2), ["application/x; a=b"]*2)
check('multipart-part-type-preserved fixture 4', solve("x/y",1), ["x/y"])
check('multipart-part-type-preserved fixture 5', solve("",1), [""])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Every part uses the selected representation media type.

Return one Content-Type value per part, preserving the full selected object type including parameters.

split: open-access


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

N = 1
observations = []
def solve(headers, payloads, closing):
    return sum(map(len,payloads))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-length-includes-framing fixture 0', solve( ["h"*N],["p"*N],"end"), 2*N+3)
check('multipart-length-includes-framing fixture 1', solve([],[],"end"), 3)
check('multipart-length-includes-framing fixture 2', solve(["ab","c"],["de","f"],"x"), 7)
check('multipart-length-includes-framing fixture 3', solve([""],[""],""), 0)
check('multipart-length-includes-framing fixture 4', solve(["h"],[""],"x"), 2)
check('multipart-length-includes-framing fixture 5', solve([""],["pp"],"x"), 3)
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(headers, payloads, closing):
    return sum(map(len,headers))+sum(map(len,payloads))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-length-includes-framing fixture 0', solve( ["h"*N],["p"*N],"end"), 2*N+3)
check('multipart-length-includes-framing fixture 1', solve([],[],"end"), 3)
check('multipart-length-includes-framing fixture 2', solve(["ab","c"],["de","f"],"x"), 7)
check('multipart-length-includes-framing fixture 3', solve([""],[""],""), 0)
check('multipart-length-includes-framing fixture 4', solve(["h"],[""],"x"), 2)
check('multipart-length-includes-framing fixture 5', solve([""],["pp"],"x"), 3)
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:
Multipart Content-Length includes headers and closing bytes.

Inputs are ASCII strings, with each header already including its separator and each payload already including its trailing framing CRLF. Return total wire length.

split: open-access


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

N = 1
observations = []
def solve(parts, individual):
    return [individual[0] if individual else None,individual]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-no-top-content-range fixture 0', solve(2,["a"+str(N),"b"]), [None,["a"+str(N),"b"]])
check('multipart-no-top-content-range fixture 1', solve(1,["a"+str(N)]), ["a"+str(N),[]])
check('multipart-no-top-content-range fixture 2', solve(0,[]), [None,[]])
check('multipart-no-top-content-range fixture 3', solve(3,["a","b","c"]), [None,["a","b","c"]])
check('multipart-no-top-content-range fixture 4', solve(1,[""]), ["",[]])
check('multipart-no-top-content-range fixture 5', solve(2,["a","a"]), [None,["a","a"]])
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(parts, individual):
    return [None,individual]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-no-top-content-range fixture 0', solve(2,["a"+str(N),"b"]), [None,["a"+str(N),"b"]])
check('multipart-no-top-content-range fixture 1', solve(1,["a"+str(N)]), ["a"+str(N),[]])
check('multipart-no-top-content-range fixture 2', solve(0,[]), [None,[]])
check('multipart-no-top-content-range fixture 3', solve(3,["a","b","c"]), [None,["a","b","c"]])
check('multipart-no-top-content-range fixture 4', solve(1,[""]), ["",[]])
check('multipart-no-top-content-range fixture 5', solve(2,["a","a"]), [None,["a","a"]])
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:
Top-level range metadata is absent for multipart bodies.

Return [top-content-range,part-ranges]. A singleton uses its individual range at top and no per-part list; multiple parts have null top and all per-part ranges.

split: open-access


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

N = 1
observations = []
def solve(parts):
    return [p[2] for p in sorted(parts,key=lambda p:p[1])]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-part-order fixture 0', solve([[1,0,"b"],[0,N,"a"]]), ["a","b"])
check('multipart-part-order fixture 1', solve([[0,N,"a"],[1,0,"b"]]), ["a","b"])
check('multipart-part-order fixture 2', solve([]), [])
check('multipart-part-order fixture 3', solve([[0,0,"x"*N]]), ["x"*N])
check('multipart-part-order fixture 4', solve([[2,0,"c"],[0,0,"a"],[1,0,"b"]]), ["a","b","c"])
check('multipart-part-order fixture 5', solve([[0,0,""],[1,1,"b"]]), ["","b"])
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(parts):
    return [p[2] for p in parts]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-part-order fixture 0', solve([[1,0,"b"],[0,N,"a"]]), ["a","b"])
check('multipart-part-order fixture 1', solve([[0,N,"a"],[1,0,"b"]]), ["a","b"])
check('multipart-part-order fixture 2', solve([]), [])
check('multipart-part-order fixture 3', solve([[0,0,"x"*N]]), ["x"*N])
check('multipart-part-order fixture 4', solve([[2,0,"c"],[0,0,"a"],[1,0,"b"]]), ["a","b","c"])
check('multipart-part-order fixture 5', solve([[0,0,""],[1,1,"b"]]), ["","b"])
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:
Multipart assembly preserves the normalized request order.

Parts are [request-position,absolute-start,payload]. Return payloads ordered by request-position, not by byte offset or arrival order. Positions unique.

split: open-access


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

N = 1
observations = []
def solve(expected_lengths, bodies):
    return "complete"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-short-read-abort fixture 0', solve([N],["x"*(N-1)]), "abort")
check('multipart-short-read-abort fixture 1', solve([N],["x"*(N+1)]), "abort")
check('multipart-short-read-abort fixture 2', solve([N],["x"*N]), "complete")
check('multipart-short-read-abort fixture 3', solve([1,1],["x"]), "abort")
check('multipart-short-read-abort fixture 4', solve([],[]), "complete")
check('multipart-short-read-abort fixture 5', solve([0],[""]), "complete")
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(expected_lengths, bodies):
    return "complete" if all(len(b)>=n for n,b in zip(expected_lengths,bodies)) else "abort"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-short-read-abort fixture 0', solve([N],["x"*(N-1)]), "abort")
check('multipart-short-read-abort fixture 1', solve([N],["x"*(N+1)]), "abort")
check('multipart-short-read-abort fixture 2', solve([N],["x"*N]), "complete")
check('multipart-short-read-abort fixture 3', solve([1,1],["x"]), "abort")
check('multipart-short-read-abort fixture 4', solve([],[]), "complete")
check('multipart-short-read-abort fixture 5', solve([0],[""]), "complete")
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 short part read prevents emission of a successful closing delimiter.

Return complete only when each part has exactly its promised ASCII byte length and the part counts match; otherwise abort. Empty part list is complete.

split: open-access


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

N = 1
observations = []
def solve(media_type):
    return media_type
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-header-injection fixture 0', solve("text/x\r\nX: "+str(N)), "application/octet-stream")
check('multipart-header-injection fixture 1', solve("text/x\nX: y"), "application/octet-stream")
check('multipart-header-injection fixture 2', solve("text/x\rX: y"), "application/octet-stream")
check('multipart-header-injection fixture 3', solve("text/plain"), "text/plain")
check('multipart-header-injection fixture 4', solve("application/x; n="+str(N)), "application/x; n="+str(N))
check('multipart-header-injection fixture 5', solve(""), "")
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(media_type):
    return media_type.replace("\r\n", " ")
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('multipart-header-injection fixture 0', solve("text/x\r\nX: "+str(N)), "application/octet-stream")
check('multipart-header-injection fixture 1', solve("text/x\nX: y"), "application/octet-stream")
check('multipart-header-injection fixture 2', solve("text/x\rX: y"), "application/octet-stream")
check('multipart-header-injection fixture 3', solve("text/plain"), "text/plain")
check('multipart-header-injection fixture 4', solve("application/x; n="+str(N)), "application/x; n="+str(N))
check('multipart-header-injection fixture 5', solve(""), "")
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:
Part media types cannot insert extra framing lines.

Return media_type if it contains neither CR nor LF; otherwise return application/octet-stream. This local serializer does not accept folded fields.

split: open-access


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

N = 1
observations = []
def solve(fragments):
    return "".join(p for tag,p in fragments)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-revision-splice fixture 0', solve([["a","x"*N],["b","y"]]), None)
check('cache-revision-splice fixture 1', solve([["a","x"],["b",""]]), None)
check('cache-revision-splice fixture 2', solve([["a","x"*N],["a","y"]]), "x"*N+"y")
check('cache-revision-splice fixture 3', solve([]), "")
check('cache-revision-splice fixture 4', solve([["a",""]]), "")
check('cache-revision-splice fixture 5', solve([["a","x"]]), "x")
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(fragments):
    return "".join(p for tag,p in fragments) if len({tag for tag,p in fragments if p})<=1 else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-revision-splice fixture 0', solve([["a","x"*N],["b","y"]]), None)
check('cache-revision-splice fixture 1', solve([["a","x"],["b",""]]), None)
check('cache-revision-splice fixture 2', solve([["a","x"*N],["a","y"]]), "x"*N+"y")
check('cache-revision-splice fixture 3', solve([]), "")
check('cache-revision-splice fixture 4', solve([["a",""]]), "")
check('cache-revision-splice fixture 5', solve([["a","x"]]), "x")
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:
Cached fragments with different validators cannot be joined.

Fragments are [strong-validator,payload] in byte order. Return their joined body when all validators agree, None otherwise; no fragments returns empty.

split: open-access


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

N = 1
observations = []
def solve(fragments):
    return sorted({s+i:c for s,b in fragments for i,c in enumerate(b)}.items())
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-overlap-byte-conflict fixture 0', solve([[N,"ab"],[N+1,"x"]]), None)
check('cache-overlap-byte-conflict fixture 1', solve([[N,"ab"],[N+1,"b"]]), [(N,"a"),(N+1,"b")])
check('cache-overlap-byte-conflict fixture 2', solve([]), [])
check('cache-overlap-byte-conflict fixture 3', solve([[0,"a"]]), [(0,"a")])
check('cache-overlap-byte-conflict fixture 4', solve([[1,"b"],[0,"a"]]), [(0,"a"),(1,"b")])
check('cache-overlap-byte-conflict fixture 5', solve([[0,""]]), [])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(fragments):
    seen={}
    for s,b in fragments:
        for i,c in enumerate(b):
            if s+i in seen: return None
            seen[s+i]=c
    return sorted(seen.items())
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-overlap-byte-conflict fixture 0', solve([[N,"ab"],[N+1,"x"]]), None)
check('cache-overlap-byte-conflict fixture 1', solve([[N,"ab"],[N+1,"b"]]), [(N,"a"),(N+1,"b")])
check('cache-overlap-byte-conflict fixture 2', solve([]), [])
check('cache-overlap-byte-conflict fixture 3', solve([[0,"a"]]), [(0,"a")])
check('cache-overlap-byte-conflict fixture 4', solve([[1,"b"],[0,"a"]]), [(0,"a"),(1,"b")])
check('cache-overlap-byte-conflict fixture 5', solve([[0,""]]), [])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Overlapping fragments must agree on their bytes.

Fragments are [absolute-offset,ASCII-body]. Return sorted [offset,char] cells; reject conflicting bytes with None. Identical overlap is permitted.

split: open-access


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

N = 1
observations = []
def solve(cells, offsets):
    return "".join(cells.get(i,"\x00") for i in offsets)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-holes-not-zero-filled fixture 0', solve( {N:"a"},[N,N+1]), None)
check('cache-holes-not-zero-filled fixture 1', solve({},[N]), None)
check('cache-holes-not-zero-filled fixture 2', solve({N:"a"},[N]), "a")
check('cache-holes-not-zero-filled fixture 3', solve({},[]), "")
check('cache-holes-not-zero-filled fixture 4', solve({0:"\x00"},[0]), "\x00")
check('cache-holes-not-zero-filled fixture 5', solve({0:"a",1:"b"},[1,0]), "ba")
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(cells, offsets):
    return "".join(cells[i] for i in offsets if i in cells)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-holes-not-zero-filled fixture 0', solve( {N:"a"},[N,N+1]), None)
check('cache-holes-not-zero-filled fixture 1', solve({},[N]), None)
check('cache-holes-not-zero-filled fixture 2', solve({N:"a"},[N]), "a")
check('cache-holes-not-zero-filled fixture 3', solve({},[]), "")
check('cache-holes-not-zero-filled fixture 4', solve({0:"\x00"},[0]), "\x00")
check('cache-holes-not-zero-filled fixture 5', solve({0:"a",1:"b"},[1,0]), "ba")
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:
Missing cached bytes remain unavailable instead of becoming zero bytes.

cells maps integer offsets to one-character strings. Return the requested bytes only if every requested offset exists, else None; empty request returns empty.

split: open-access


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

N = 1
observations = []
def solve(known, incoming, body):
    return [incoming,body]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-total-length-disagreement fixture 0', solve(N,N+1,"p"*N), None)
check('cache-total-length-disagreement fixture 1', solve(N,None,"p"), [N,"p"])
check('cache-total-length-disagreement fixture 2', solve(None,N,"p"), [N,"p"])
check('cache-total-length-disagreement fixture 3', solve(None,None,""), [None,""])
check('cache-total-length-disagreement fixture 4', solve(0,0,""), [0,""])
check('cache-total-length-disagreement fixture 5', solve(N,N,"p"*N), [N,"p"*N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(known, incoming, body):
    return [max(known or 0,incoming or 0),body]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-total-length-disagreement fixture 0', solve(N,N+1,"p"*N), None)
check('cache-total-length-disagreement fixture 1', solve(N,None,"p"), [N,"p"])
check('cache-total-length-disagreement fixture 2', solve(None,N,"p"), [N,"p"])
check('cache-total-length-disagreement fixture 3', solve(None,None,""), [None,""])
check('cache-total-length-disagreement fixture 4', solve(0,0,""), [0,""])
check('cache-total-length-disagreement fixture 5', solve(N,N,"p"*N), [N,"p"*N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Fragment admission rejects a conflicting complete length.

Known and incoming complete lengths are integer or None. Reject conflicting known lengths; otherwise return [resolved-length,body].

split: open-access


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

N = 1
observations = []
def solve(ages, elapsed):
    return min(ages,default=0)+elapsed
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-age-oldest-fragment fixture 0', solve([N,3*N],N), 4*N)
check('cache-age-oldest-fragment fixture 1', solve([3*N,N],0), 3*N)
check('cache-age-oldest-fragment fixture 2', solve([],N), N)
check('cache-age-oldest-fragment fixture 3', solve([0,0],N), N)
check('cache-age-oldest-fragment fixture 4', solve([N],N), 2*N)
check('cache-age-oldest-fragment fixture 5', solve([0,N,0],0), N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(ages, elapsed):
    return sum(ages)//len(ages)+elapsed if ages else elapsed
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-age-oldest-fragment fixture 0', solve([N,3*N],N), 4*N)
check('cache-age-oldest-fragment fixture 1', solve([3*N,N],0), 3*N)
check('cache-age-oldest-fragment fixture 2', solve([],N), N)
check('cache-age-oldest-fragment fixture 3', solve([0,0],N), N)
check('cache-age-oldest-fragment fixture 4', solve([N],N), 2*N)
check('cache-age-oldest-fragment fixture 5', solve([0,N,0],0), N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A stitched response exposes the oldest contributing fragment age.

Return max fragment age plus elapsed, or elapsed for no fragments. Inputs nonnegative integer seconds.

split: open-access


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

N = 1
observations = []
def solve(expiries, now):
    return max(0,max(expiries,default=now)-now)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-expiry-minimum fixture 0', solve([N,3*N],0), N)
check('cache-expiry-minimum fixture 1', solve([N,3*N],2*N), 0)
check('cache-expiry-minimum fixture 2', solve([],N), 0)
check('cache-expiry-minimum fixture 3', solve([N],0), N)
check('cache-expiry-minimum fixture 4', solve([N,N],N), 0)
check('cache-expiry-minimum fixture 5', solve([2*N,4*N],N), N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(expiries, now):
    return max(0,sum(expiries)//len(expiries)-now) if expiries else 0
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-expiry-minimum fixture 0', solve([N,3*N],0), N)
check('cache-expiry-minimum fixture 1', solve([N,3*N],2*N), 0)
check('cache-expiry-minimum fixture 2', solve([],N), 0)
check('cache-expiry-minimum fixture 3', solve([N],0), N)
check('cache-expiry-minimum fixture 4', solve([N,N],N), 0)
check('cache-expiry-minimum fixture 5', solve([2*N,4*N],N), N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Assembly expires with its earliest expiring fragment.

Return remaining freshness min(expiries)-now clipped at zero, or zero when no fragments contribute.

split: open-access


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

N = 1
observations = []
def solve(url, private, principal, validator):
    return [url,validator,None]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-private-partition fixture 0', solve("/x",True,"u"+str(N),"v"), ["/x","v","u"+str(N)])
check('cache-private-partition fixture 1', solve("/x",True,"","v"), ["/x","v",""])
check('cache-private-partition fixture 2', solve("/x",False,"u","v"), ["/x","v",None])
check('cache-private-partition fixture 3', solve("/y",False,"","w"), ["/y","w",None])
check('cache-private-partition fixture 4', solve("/x",True,"a","w"), ["/x","w","a"])
check('cache-private-partition fixture 5', solve("/x",True,"b","w"), ["/x","w","b"])
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(url, private, principal, validator):
    return [url,validator,principal or None]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-private-partition fixture 0', solve("/x",True,"u"+str(N),"v"), ["/x","v","u"+str(N)])
check('cache-private-partition fixture 1', solve("/x",True,"","v"), ["/x","v",""])
check('cache-private-partition fixture 2', solve("/x",False,"u","v"), ["/x","v",None])
check('cache-private-partition fixture 3', solve("/y",False,"","w"), ["/y","w",None])
check('cache-private-partition fixture 4', solve("/x",True,"a","w"), ["/x","w","a"])
check('cache-private-partition fixture 5', solve("/x",True,"b","w"), ["/x","w","b"])
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:
User-specific partial objects remain partitioned by principal.

Return cache identity [url,validator,principal-or-null]; principal participates only for private responses, including empty principal.

split: open-access


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

N = 1
observations = []
def solve(vary, payload):
    return ["store",payload]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-vary-star-bypass fixture 0', solve( ["accept-encoding","*"],"p"*N), ["bypass",""])
check('cache-vary-star-bypass fixture 1', solve([" * "],"p"), ["bypass",""])
check('cache-vary-star-bypass fixture 2', solve(["*"],"p"), ["bypass",""])
check('cache-vary-star-bypass fixture 3', solve([],"p"*N), ["store","p"*N])
check('cache-vary-star-bypass fixture 4', solve(["x-star"],"p"), ["store","p"])
check('cache-vary-star-bypass fixture 5', solve(["accept-language"],""), ["store",""])
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(vary, payload):
    return ["bypass",""] if vary==["*"] else ["store",payload]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-vary-star-bypass fixture 0', solve( ["accept-encoding","*"],"p"*N), ["bypass",""])
check('cache-vary-star-bypass fixture 1', solve([" * "],"p"), ["bypass",""])
check('cache-vary-star-bypass fixture 2', solve(["*"],"p"), ["bypass",""])
check('cache-vary-star-bypass fixture 3', solve([],"p"*N), ["store","p"*N])
check('cache-vary-star-bypass fixture 4', solve(["x-star"],"p"), ["store","p"])
check('cache-vary-star-bypass fixture 5', solve(["accept-language"],""), ["store",""])
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 wildcard Vary response is not stored as a reusable partial object.

Return [store,payload] unless parsed Vary tokens include *, then [bypass,empty]. Tokens may have surrounding ASCII space.

split: open-access


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

N = 1
observations = []
def solve(old, full):
    r=dict(old)
    r.update(enumerate(full))
    return sorted(r.items())
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-full-response-replaces-fragments fixture 0', solve( {0:"x",N+2:"z"},"a"*N), list(enumerate("a"*N)))
check('cache-full-response-replaces-fragments fixture 1', solve({0:"x"},""), [])
check('cache-full-response-replaces-fragments fixture 2', solve({},"ab"), [(0,"a"),(1,"b")])
check('cache-full-response-replaces-fragments fixture 3', solve({0:"x"},"a"), [(0,"a")])
check('cache-full-response-replaces-fragments fixture 4', solve({},""), [])
check('cache-full-response-replaces-fragments fixture 5', solve({1:"b"},"xy"), [(0,"x"),(1,"y")])
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(old, full):
    return sorted(old.items()) if not full else list(enumerate(full))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-full-response-replaces-fragments fixture 0', solve( {0:"x",N+2:"z"},"a"*N), list(enumerate("a"*N)))
check('cache-full-response-replaces-fragments fixture 1', solve({0:"x"},""), [])
check('cache-full-response-replaces-fragments fixture 2', solve({},"ab"), [(0,"a"),(1,"b")])
check('cache-full-response-replaces-fragments fixture 3', solve({0:"x"},"a"), [(0,"a")])
check('cache-full-response-replaces-fragments fixture 4', solve({},""), [])
check('cache-full-response-replaces-fragments fixture 5', solve({1:"b"},"xy"), [(0,"x"),(1,"y")])
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 fresh full response replaces all stale sparse fragments.

old maps absolute offsets to old chars. A completed full ASCII body atomically replaces the whole partial cache, including removal of bytes past new EOF.

split: open-access


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

N = 1
observations = []
def solve(no_store, staged, committed):
    return [staged,committed]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-no-store-purges-staging fixture 0', solve(True,"s"*N,"c"*N), ["",""])
check('cache-no-store-purges-staging fixture 1', solve(True,"s",""), ["",""])
check('cache-no-store-purges-staging fixture 2', solve(True,"","c"), ["",""])
check('cache-no-store-purges-staging fixture 3', solve(False,"s"*N,"c"), ["s"*N,"c"])
check('cache-no-store-purges-staging fixture 4', solve(False,"",""), ["",""])
check('cache-no-store-purges-staging fixture 5', solve(False,"s",""), ["s",""])
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(no_store, staged, committed):
    return [staged,""] if no_store else [staged,committed]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('cache-no-store-purges-staging fixture 0', solve(True,"s"*N,"c"*N), ["",""])
check('cache-no-store-purges-staging fixture 1', solve(True,"s",""), ["",""])
check('cache-no-store-purges-staging fixture 2', solve(True,"","c"), ["",""])
check('cache-no-store-purges-staging fixture 3', solve(False,"s"*N,"c"), ["s"*N,"c"])
check('cache-no-store-purges-staging fixture 4', solve(False,"",""), ["",""])
check('cache-no-store-purges-staging fixture 5', solve(False,"s",""), ["s",""])
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:
No-store metadata prevents partial staging from surviving admission.

Return [staged,committed] unchanged for reusable responses; no-store clears both temporary and committed bytes for this object.

split: open-access


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

N = 1
observations = []
def solve(pending, accepted):
    return ""
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-short-write-resume fixture 0', solve("abc"*N,N), ("abc"*N)[N:])
check('stream-short-write-resume fixture 1', solve("abc",0), "abc")
check('stream-short-write-resume fixture 2', solve("abc",3), "")
check('stream-short-write-resume fixture 3', solve("",0), "")
check('stream-short-write-resume fixture 4', solve("x",1), "")
check('stream-short-write-resume fixture 5', solve("abcd",2), "cd")
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(pending, accepted):
    return pending[accepted:] if accepted else ""
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-short-write-resume fixture 0', solve("abc"*N,N), ("abc"*N)[N:])
check('stream-short-write-resume fixture 1', solve("abc",0), "abc")
check('stream-short-write-resume fixture 2', solve("abc",3), "")
check('stream-short-write-resume fixture 3', solve("",0), "")
check('stream-short-write-resume fixture 4', solve("x",1), "")
check('stream-short-write-resume fixture 5', solve("abcd",2), "cd")
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 short socket write preserves the unsent byte suffix.

accepted is a valid integer count in [0,len(pending)]. Return remaining ASCII bytes after only that count was accepted.

split: open-access


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

N = 1
observations = []
def solve(credit, remaining, block):
    return min(remaining,block)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-credit-backpressure fixture 0', solve(N,3*N,2*N), N)
check('stream-credit-backpressure fixture 1', solve(3*N,N,2*N), N)
check('stream-credit-backpressure fixture 2', solve(3*N,3*N,N), N)
check('stream-credit-backpressure fixture 3', solve(0,N,N), 0)
check('stream-credit-backpressure fixture 4', solve(N,0,N), 0)
check('stream-credit-backpressure fixture 5', solve(N,N,0), 0)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(credit, remaining, block):
    return min(credit,block)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-credit-backpressure fixture 0', solve(N,3*N,2*N), N)
check('stream-credit-backpressure fixture 1', solve(3*N,N,2*N), N)
check('stream-credit-backpressure fixture 2', solve(3*N,3*N,N), N)
check('stream-credit-backpressure fixture 3', solve(0,N,N), 0)
check('stream-credit-backpressure fixture 4', solve(N,0,N), 0)
check('stream-credit-backpressure fixture 5', solve(N,N,0), 0)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Read-ahead respects both stream credit and remaining range bytes.

Return the next permitted read size min(credit,remaining,block), all nonnegative.

split: open-access


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

N = 1
observations = []
def solve(start, delivered):
    return delivered
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-progress-absolute-offset fixture 0', solve(2*N,N), 3*N)
check('stream-progress-absolute-offset fixture 1', solve(N,0), N)
check('stream-progress-absolute-offset fixture 2', solve(0,N), N)
check('stream-progress-absolute-offset fixture 3', solve(0,0), 0)
check('stream-progress-absolute-offset fixture 4', solve(N,2*N), 3*N)
check('stream-progress-absolute-offset fixture 5', solve(3*N,4*N), 7*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(start, delivered):
    return max(start,delivered)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-progress-absolute-offset fixture 0', solve(2*N,N), 3*N)
check('stream-progress-absolute-offset fixture 1', solve(N,0), N)
check('stream-progress-absolute-offset fixture 2', solve(0,N), N)
check('stream-progress-absolute-offset fixture 3', solve(0,0), 0)
check('stream-progress-absolute-offset fixture 4', solve(N,2*N), 3*N)
check('stream-progress-absolute-offset fixture 5', solve(3*N,4*N), 7*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Partial-read progress advances from the selected absolute start.

Return the next absolute representation offset start+delivered; delivered counts body bytes only.

split: open-access


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

N = 1
observations = []
def solve(offset, payload_written, framing_written):
    return offset+payload_written+framing_written
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-framing-not-progress fixture 0', solve(N,N,3*N), 2*N)
check('stream-framing-not-progress fixture 1', solve(N,0,N), N)
check('stream-framing-not-progress fixture 2', solve(N,N,0), 2*N)
check('stream-framing-not-progress fixture 3', solve(0,0,0), 0)
check('stream-framing-not-progress fixture 4', solve(0,2*N,N), 2*N)
check('stream-framing-not-progress fixture 5', solve(3*N,N,N), 4*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(offset, payload_written, framing_written):
    return offset+max(payload_written,framing_written)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-framing-not-progress fixture 0', solve(N,N,3*N), 2*N)
check('stream-framing-not-progress fixture 1', solve(N,0,N), N)
check('stream-framing-not-progress fixture 2', solve(N,N,0), 2*N)
check('stream-framing-not-progress fixture 3', solve(0,0,0), 0)
check('stream-framing-not-progress fixture 4', solve(0,2*N,N), 2*N)
check('stream-framing-not-progress fixture 5', solve(3*N,N,N), 4*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Multipart framing bytes do not advance the representation cursor.

Return representation cursor after payload writes; framing bytes are tracked by the transport separately.

split: open-access


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

N = 1
observations = []
def solve(remaining, eof):
    return "complete" if eof else "reading"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-premature-eof-state fixture 0', solve(N,True), "truncated")
check('stream-premature-eof-state fixture 1', solve(0,True), "complete")
check('stream-premature-eof-state fixture 2', solve(0,False), "complete")
check('stream-premature-eof-state fixture 3', solve(N,False), "reading")
check('stream-premature-eof-state fixture 4', solve(2*N,True), "truncated")
check('stream-premature-eof-state fixture 5', solve(2*N,False), "reading")
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(remaining, eof):
    return "truncated" if eof else ("complete" if remaining==0 else "reading")
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-premature-eof-state fixture 0', solve(N,True), "truncated")
check('stream-premature-eof-state fixture 1', solve(0,True), "complete")
check('stream-premature-eof-state fixture 2', solve(0,False), "complete")
check('stream-premature-eof-state fixture 3', solve(N,False), "reading")
check('stream-premature-eof-state fixture 4', solve(2*N,True), "truncated")
check('stream-premature-eof-state fixture 5', solve(2*N,False), "reading")
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
EOF before the promised range length marks the stream truncated.

Return truncated for EOF with remaining>0, complete for zero remaining, otherwise reading.

split: open-access


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

N = 1
observations = []
def solve(data, eof, blocked):
    return ["data",data] if data else ["eof",""]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-empty-read-would-block fixture 0', solve("",False,True), ["wait",""])
check('stream-empty-read-would-block fixture 1', solve("",False,False), ["idle",""])
check('stream-empty-read-would-block fixture 2', solve("",True,False), ["eof",""])
check('stream-empty-read-would-block fixture 3', solve("p"*N,False,False), ["data","p"*N])
check('stream-empty-read-would-block fixture 4', solve("x",True,False), ["data","x"])
check('stream-empty-read-would-block fixture 5', solve("p"*N,False,True), ["data","p"*N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(data, eof, blocked):
    return ["data",data] if data else (["wait",""] if blocked else ["eof",""])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-empty-read-would-block fixture 0', solve("",False,True), ["wait",""])
check('stream-empty-read-would-block fixture 1', solve("",False,False), ["idle",""])
check('stream-empty-read-would-block fixture 2', solve("",True,False), ["eof",""])
check('stream-empty-read-would-block fixture 3', solve("p"*N,False,False), ["data","p"*N])
check('stream-empty-read-would-block fixture 4', solve("x",True,False), ["data","x"])
check('stream-empty-read-would-block fixture 5', solve("p"*N,False,True), ["data","p"*N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A would-block read does not mean representation EOF.

Return data if nonempty; otherwise wait when blocked, eof when explicit EOF, else idle. The transport never marks blocked and eof simultaneously.

split: open-access


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

N = 1
observations = []
def solve(cancelled, pins, pending):
    return [pins,""] if cancelled else [pins,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-cancel-releases-pin fixture 0', solve(True,1,"p"*N), [0,""])
check('stream-cancel-releases-pin fixture 1', solve(True,0,"p"), [0,""])
check('stream-cancel-releases-pin fixture 2', solve(True,1,""), [0,""])
check('stream-cancel-releases-pin fixture 3', solve(False,1,"p"*N), [1,"p"*N])
check('stream-cancel-releases-pin fixture 4', solve(False,0,""), [0,""])
check('stream-cancel-releases-pin fixture 5', solve(False,0,"p"), [0,"p"])
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(cancelled, pins, pending):
    return [0,pending] if cancelled else [pins,pending]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-cancel-releases-pin fixture 0', solve(True,1,"p"*N), [0,""])
check('stream-cancel-releases-pin fixture 1', solve(True,0,"p"), [0,""])
check('stream-cancel-releases-pin fixture 2', solve(True,1,""), [0,""])
check('stream-cancel-releases-pin fixture 3', solve(False,1,"p"*N), [1,"p"*N])
check('stream-cancel-releases-pin fixture 4', solve(False,0,""), [0,""])
check('stream-cancel-releases-pin fixture 5', solve(False,0,"p"), [0,"p"])
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:
Canceled range streams release snapshot pins and pending data.

pins is the current stream-owned count (0 or1); cancellation clears both pin and pending body, otherwise preserves state.

split: open-access


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

N = 1
observations = []
def solve(events):
    return [events.count("close"),"close" not in events]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-close-idempotency fixture 0', solve( ["close"]*(N+1)), [1,False])
check('stream-close-idempotency fixture 1', solve([]), [0,True])
check('stream-close-idempotency fixture 2', solve(["tick"]*N), [0,True])
check('stream-close-idempotency fixture 3', solve(["close"]), [1,False])
check('stream-close-idempotency fixture 4', solve(["tick","close","tick"]), [1,False])
check('stream-close-idempotency fixture 5', solve(["close","tick","close"]), [1,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(events):
    return [1,False]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-close-idempotency fixture 0', solve( ["close"]*(N+1)), [1,False])
check('stream-close-idempotency fixture 1', solve([]), [0,True])
check('stream-close-idempotency fixture 2', solve(["tick"]*N), [0,True])
check('stream-close-idempotency fixture 3', solve(["close"]), [1,False])
check('stream-close-idempotency fixture 4', solve(["tick","close","tick"]), [1,False])
check('stream-close-idempotency fixture 5', solve(["close","tick","close"]), [1,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:
Repeated close events release a range handle only once.

Events are close or tick. Initially one handle is open. Return [release-count,open]; subsequent close is a no-op.

split: open-access


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

N = 1
observations = []
def solve(payloads, headers):
    return sum(map(ord,"".join(payloads+headers)))%257
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-checksum-body-only fixture 0', solve( ["z"*(N+3)],["Header"]), (122*(N+3))%257)
check('stream-checksum-body-only fixture 1', solve(["a"],["b"]), 97)
check('stream-checksum-body-only fixture 2', solve([],["header"]), 0)
check('stream-checksum-body-only fixture 3', solve([],[]), 0)
check('stream-checksum-body-only fixture 4', solve(["ab"],[]), 195)
check('stream-checksum-body-only fixture 5', solve(["z","z","z"],[]), 109)
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(payloads, headers):
    return sum(map(ord,"".join(payloads)))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-checksum-body-only fixture 0', solve( ["z"*(N+3)],["Header"]), (122*(N+3))%257)
check('stream-checksum-body-only fixture 1', solve(["a"],["b"]), 97)
check('stream-checksum-body-only fixture 2', solve([],["header"]), 0)
check('stream-checksum-body-only fixture 3', solve([],[]), 0)
check('stream-checksum-body-only fixture 4', solve(["ab"],[]), 195)
check('stream-checksum-body-only fixture 5', solve(["z","z","z"],[]), 109)
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:
Representation checksum excludes multipart envelope bytes.

Local checksum is sum ASCII code points modulo 257 over payloads only. Framing headers are not representation data.

split: open-access


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

N = 1
observations = []
def solve(headers_sent, failure, body):
    return ["error500",""] if failure else ["continue",body]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-error-after-headers fixture 0', solve(True,True,"p"*N), ["abort",""])
check('stream-error-after-headers fixture 1', solve(True,False,"p"*N), ["continue","p"*N])
check('stream-error-after-headers fixture 2', solve(False,True,"p"), ["error500",""])
check('stream-error-after-headers fixture 3', solve(False,False,"p"), ["continue","p"])
check('stream-error-after-headers fixture 4', solve(True,True,""), ["abort",""])
check('stream-error-after-headers fixture 5', solve(True,False,""), ["continue",""])
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(headers_sent, failure, body):
    return ["abort",""] if headers_sent else (["error500",""] if failure else ["continue",body])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-error-after-headers fixture 0', solve(True,True,"p"*N), ["abort",""])
check('stream-error-after-headers fixture 1', solve(True,False,"p"*N), ["continue","p"*N])
check('stream-error-after-headers fixture 2', solve(False,True,"p"), ["error500",""])
check('stream-error-after-headers fixture 3', solve(False,False,"p"), ["continue","p"])
check('stream-error-after-headers fixture 4', solve(True,True,""), ["abort",""])
check('stream-error-after-headers fixture 5', solve(True,False,""), ["continue",""])
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:
Errors after partial headers abort instead of injecting a second response.

If failure occurs after headers, abort with empty body; before headers return error500. No failure continues body.

split: open-access


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

N = 1
observations = []
def solve(remaining, aborted, digest):
    return digest
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-trailer-after-completion fixture 0', solve(N,False,"d"+str(N)), None)
check('stream-trailer-after-completion fixture 1', solve(0,True,"d"), None)
check('stream-trailer-after-completion fixture 2', solve(0,False,"d"+str(N)), "d"+str(N))
check('stream-trailer-after-completion fixture 3', solve(N,True,"d"), None)
check('stream-trailer-after-completion fixture 4', solve(0,False,""), "")
check('stream-trailer-after-completion fixture 5', solve(0,True,""), 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(remaining, aborted, digest):
    return digest if remaining==0 else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-trailer-after-completion fixture 0', solve(N,False,"d"+str(N)), None)
check('stream-trailer-after-completion fixture 1', solve(0,True,"d"), None)
check('stream-trailer-after-completion fixture 2', solve(0,False,"d"+str(N)), "d"+str(N))
check('stream-trailer-after-completion fixture 3', solve(N,True,"d"), None)
check('stream-trailer-after-completion fixture 4', solve(0,False,""), "")
check('stream-trailer-after-completion fixture 5', solve(0,True,""), 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:
Integrity trailers are emitted only after all promised bytes arrive.

Return digest only after nonaborted completion, None while incomplete or aborted.

split: open-access


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

N = 1
observations = []
def solve(data, remaining):
    return [data,""]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-read-result-cap fixture 0', solve("abc"*N,N), [("abc"*N)[:N],("abc"*N)[N:]])
check('stream-read-result-cap fixture 1', solve("abc",0), ["","abc"])
check('stream-read-result-cap fixture 2', solve("abc",3), ["abc",""])
check('stream-read-result-cap fixture 3', solve("abc",5), ["abc",""])
check('stream-read-result-cap fixture 4', solve("",N), ["",""])
check('stream-read-result-cap fixture 5', solve("abcd",2), ["ab","cd"])
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(data, remaining):
    return [data[:remaining],""]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-read-result-cap fixture 0', solve("abc"*N,N), [("abc"*N)[:N],("abc"*N)[N:]])
check('stream-read-result-cap fixture 1', solve("abc",0), ["","abc"])
check('stream-read-result-cap fixture 2', solve("abc",3), ["abc",""])
check('stream-read-result-cap fixture 3', solve("abc",5), ["abc",""])
check('stream-read-result-cap fixture 4', solve("",N), ["",""])
check('stream-read-result-cap fixture 5', solve("abcd",2), ["ab","cd"])
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:
Oversized backend reads do not leak bytes outside the response budget.

Backend may return more ASCII bytes than requested. Emit at most remaining bytes and return [emitted,unconsumed], preserving the suffix.

split: open-access


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

N = 1
observations = []
def solve(current, target, length):
    return [["read",length]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-seek-before-read fixture 0', solve(0,N,N), [["seek",N],["read",N]])
check('stream-seek-before-read fixture 1', solve(N,N,N), [["read",N]])
check('stream-seek-before-read fixture 2', solve(N,0,N), [["seek",0],["read",N]])
check('stream-seek-before-read fixture 3', solve(0,0,0), [["read",0]])
check('stream-seek-before-read fixture 4', solve(0,N,0), [["seek",N],["read",0]])
check('stream-seek-before-read fixture 5', solve(2*N,N,1), [["seek",N],["read",1]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(current, target, length):
    return [["seek",target],["read",length]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-seek-before-read fixture 0', solve(0,N,N), [["seek",N],["read",N]])
check('stream-seek-before-read fixture 1', solve(N,N,N), [["read",N]])
check('stream-seek-before-read fixture 2', solve(N,0,N), [["seek",0],["read",N]])
check('stream-seek-before-read fixture 3', solve(0,0,0), [["read",0]])
check('stream-seek-before-read fixture 4', solve(0,N,0), [["seek",N],["read",0]])
check('stream-seek-before-read fixture 5', solve(2*N,N,1), [["seek",N],["read",1]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Each multipart range begins with its own absolute seek.

Return ordered abstract I/O operations: seek target if cursor differs, then read length even for zero length; equal cursor avoids redundant seek.

split: open-access


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

N = 1
observations = []
def solve(seek_ok, target, length):
    return ["read",target,length]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-seek-failure-stops-read fixture 0', solve(False,N,N), ["abort"])
check('stream-seek-failure-stops-read fixture 1', solve(False,0,N), ["abort"])
check('stream-seek-failure-stops-read fixture 2', solve(True,N,N), ["read",N,N])
check('stream-seek-failure-stops-read fixture 3', solve(True,0,0), ["read",0,0])
check('stream-seek-failure-stops-read fixture 4', solve(True,2*N,1), ["read",2*N,1])
check('stream-seek-failure-stops-read fixture 5', solve(False,N,0), ["abort"])
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(seek_ok, target, length):
    return ["read",target if seek_ok else 0,length]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-seek-failure-stops-read fixture 0', solve(False,N,N), ["abort"])
check('stream-seek-failure-stops-read fixture 1', solve(False,0,N), ["abort"])
check('stream-seek-failure-stops-read fixture 2', solve(True,N,N), ["read",N,N])
check('stream-seek-failure-stops-read fixture 3', solve(True,0,0), ["read",0,0])
check('stream-seek-failure-stops-read fixture 4', solve(True,2*N,1), ["read",2*N,1])
check('stream-seek-failure-stops-read fixture 5', solve(False,N,0), ["abort"])
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:
Failed absolute seek prevents a range read from the stale cursor.

A failed seek yields only an abort operation. Successful seek schedules [read,target,length].

split: open-access


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

N = 1
observations = []
def solve(writes):
    return sum(a for a,b in writes)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-delivery-count-accepted fixture 0', solve([[3*N,N]]), N)
check('stream-delivery-count-accepted fixture 1', solve([[N,0],[N,N]]), N)
check('stream-delivery-count-accepted fixture 2', solve([]), 0)
check('stream-delivery-count-accepted fixture 3', solve([[N,N]]), N)
check('stream-delivery-count-accepted fixture 4', solve([[0,0]]), 0)
check('stream-delivery-count-accepted fixture 5', solve([[4*N,N],[2*N,N]]), 2*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(writes):
    return sum(b for a,b in writes if b==a)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-delivery-count-accepted fixture 0', solve([[3*N,N]]), N)
check('stream-delivery-count-accepted fixture 1', solve([[N,0],[N,N]]), N)
check('stream-delivery-count-accepted fixture 2', solve([]), 0)
check('stream-delivery-count-accepted fixture 3', solve([[N,N]]), N)
check('stream-delivery-count-accepted fixture 4', solve([[0,0]]), 0)
check('stream-delivery-count-accepted fixture 5', solve([[4*N,N],[2*N,N]]), 2*N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Download accounting counts accepted payload bytes instead of attempted bytes.

writes are [attempted,accepted] payload counts, 0<=accepted<=attempted. Return delivered count.

split: open-access


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

N = 1
observations = []
def solve(queued, new):
    return ([new] if new else [])+queued
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-buffer-retains-order fixture 0', solve( ["a"*N],"b"*N), ["a"*N,"b"*N])
check('stream-buffer-retains-order fixture 1', solve(["a"],""), ["a"])
check('stream-buffer-retains-order fixture 2', solve([],"x"), ["x"])
check('stream-buffer-retains-order fixture 3', solve([],""), [])
check('stream-buffer-retains-order fixture 4', solve(["a","b"],"c"), ["a","b","c"])
check('stream-buffer-retains-order fixture 5', solve([""],"x"), ["","x"])
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(queued, new):
    return queued+[new]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-buffer-retains-order fixture 0', solve( ["a"*N],"b"*N), ["a"*N,"b"*N])
check('stream-buffer-retains-order fixture 1', solve(["a"],""), ["a"])
check('stream-buffer-retains-order fixture 2', solve([],"x"), ["x"])
check('stream-buffer-retains-order fixture 3', solve([],""), [])
check('stream-buffer-retains-order fixture 4', solve(["a","b"],"c"), ["a","b","c"])
check('stream-buffer-retains-order fixture 5', solve([""],"x"), ["","x"])
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:
Queued partial payloads preserve FIFO order under backpressure.

Append new complete byte chunk to FIFO queue; empty chunks carry no payload and are not added.

split: open-access


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

N = 1
observations = []
def solve(last, now, delivered, timeout):
    return now+timeout
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-timeout-last-progress fixture 0', solve(N,3*N,0,N), 2*N)
check('stream-timeout-last-progress fixture 1', solve(N,3*N,N,N), 4*N)
check('stream-timeout-last-progress fixture 2', solve(0,0,0,N), N)
check('stream-timeout-last-progress fixture 3', solve(0,N,1,0), N)
check('stream-timeout-last-progress fixture 4', solve(N,N,0,N), 2*N)
check('stream-timeout-last-progress fixture 5', solve(0,N,0,0), 0)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(last, now, delivered, timeout):
    return last+timeout
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-timeout-last-progress fixture 0', solve(N,3*N,0,N), 2*N)
check('stream-timeout-last-progress fixture 1', solve(N,3*N,N,N), 4*N)
check('stream-timeout-last-progress fixture 2', solve(0,0,0,N), N)
check('stream-timeout-last-progress fixture 3', solve(0,N,1,0), N)
check('stream-timeout-last-progress fixture 4', solve(N,N,0,N), 2*N)
check('stream-timeout-last-progress fixture 5', solve(0,N,0,0), 0)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Range stream idle deadlines advance only on actual byte progress.

Return next idle deadline; positive delivered resets progress to now, zero-byte activity retains last. All integers nonnegative.

split: open-access


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

N = 1
observations = []
def solve(body, requests):
    cursor=0
    out=[]
    for start,count in requests:
        out.append(body[cursor:cursor+count]); cursor+=count
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-shared-handle-cursor fixture 0', solve("abcdef"*N,[[2,2],[0,2]]), ["cd","ab"])
check('stream-shared-handle-cursor fixture 1', solve("abcdef",[[0,1],[3,1]]), ["a","d"])
check('stream-shared-handle-cursor fixture 2', solve("",[]), [])
check('stream-shared-handle-cursor fixture 3', solve("abc",[[1,1]]), ["b"])
check('stream-shared-handle-cursor fixture 4', solve("abc",[[0,0],[2,1]]), ["","c"])
check('stream-shared-handle-cursor fixture 5', solve("abcdef",[[1,2],[1,2]]), ["bc","bc"])
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(body, requests):
    cursor=requests[0][0] if requests else 0
    out=[]
    for start,count in requests:
        out.append(body[cursor:cursor+count]); cursor+=count
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-shared-handle-cursor fixture 0', solve("abcdef"*N,[[2,2],[0,2]]), ["cd","ab"])
check('stream-shared-handle-cursor fixture 1', solve("abcdef",[[0,1],[3,1]]), ["a","d"])
check('stream-shared-handle-cursor fixture 2', solve("",[]), [])
check('stream-shared-handle-cursor fixture 3', solve("abc",[[1,1]]), ["b"])
check('stream-shared-handle-cursor fixture 4', solve("abc",[[0,0],[2,1]]), ["","c"])
check('stream-shared-handle-cursor fixture 5', solve("abcdef",[[1,2],[1,2]]), ["bc","bc"])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent readers use positional I/O instead of a shared mutable seek cursor.

requests are [absolute-offset,count]; return independent ASCII slices for each request. The underlying shared handle cursor is irrelevant.

split: open-access


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

N = 1
observations = []
def solve(completed, total, cancelled):
    return "release" if completed>0 or cancelled else "retain"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-release-after-last-part fixture 0', solve(1,N+1,False), "retain")
check('stream-release-after-last-part fixture 1', solve(0,N,True), "release")
check('stream-release-after-last-part fixture 2', solve(N,N,False), "release")
check('stream-release-after-last-part fixture 3', solve(0,0,False), "release")
check('stream-release-after-last-part fixture 4', solve(0,N,False), "retain")
check('stream-release-after-last-part fixture 5', solve(N,N,True), "release")
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(completed, total, cancelled):
    return "release" if completed==total else "retain"
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-release-after-last-part fixture 0', solve(1,N+1,False), "retain")
check('stream-release-after-last-part fixture 1', solve(0,N,True), "release")
check('stream-release-after-last-part fixture 2', solve(N,N,False), "release")
check('stream-release-after-last-part fixture 3', solve(0,0,False), "release")
check('stream-release-after-last-part fixture 4', solve(0,N,False), "retain")
check('stream-release-after-last-part fixture 5', solve(N,N,True), "release")
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:
Multipart source handles remain open until every part finishes.

Return release on cancellation or when completed equals total; otherwise retain. Valid completed in[0,total].

split: open-access


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

N = 1
observations = []
def solve(reset, queue, credit):
    return [queue,0] if reset else [queue,credit]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-reset-discards-queued-data fixture 0', solve(True,["x"*N],N), [[],0])
check('stream-reset-discards-queued-data fixture 1', solve(True,[],N), [[],0])
check('stream-reset-discards-queued-data fixture 2', solve(True,["x"],0), [[],0])
check('stream-reset-discards-queued-data fixture 3', solve(False,["x"*N],N), [["x"*N],N])
check('stream-reset-discards-queued-data fixture 4', solve(False,[],0), [[],0])
check('stream-reset-discards-queued-data fixture 5', solve(False,[],N), [[],N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(reset, queue, credit):
    return [[],credit] if reset else [queue,credit]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('stream-reset-discards-queued-data fixture 0', solve(True,["x"*N],N), [[],0])
check('stream-reset-discards-queued-data fixture 1', solve(True,[],N), [[],0])
check('stream-reset-discards-queued-data fixture 2', solve(True,["x"],0), [[],0])
check('stream-reset-discards-queued-data fixture 3', solve(False,["x"*N],N), [["x"*N],N])
check('stream-reset-discards-queued-data fixture 4', solve(False,[],0), [[],0])
check('stream-reset-discards-queued-data fixture 5', solve(False,[],N), [[],N])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A transport reset cannot leave queued range bytes available for transmission.

Reset returns [empty-queue,zero-credit]; normal operation preserves both.

split: open-access


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

N = 1
observations = []
def solve(fields):
    return fields[0] if fields else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-field-combination fixture 0', solve([["bytes",["0-"+str(N)]],["bytes",["8-9"]]]), ["bytes",["0-"+str(N),"8-9"]])
check('range-field-combination fixture 1', solve([["bytes",["0-1"]],["items",["2-3"]]]), None)
check('range-field-combination fixture 2', solve([]), None)
check('range-field-combination fixture 3', solve([["bytes",["0-1"]]]), ["bytes",["0-1"]])
check('range-field-combination fixture 4', solve([["items",["a"]],["items",["b"]]]), ["items",["a","b"]])
check('range-field-combination fixture 5', solve([["bytes",[]],["bytes",["0-0"]]]), ["bytes",["0-0"]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(fields):
    return [fields[0][0],[m for u,ms in fields for m in ms]] if fields else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-field-combination fixture 0', solve([["bytes",["0-"+str(N)]],["bytes",["8-9"]]]), ["bytes",["0-"+str(N),"8-9"]])
check('range-field-combination fixture 1', solve([["bytes",["0-1"]],["items",["2-3"]]]), None)
check('range-field-combination fixture 2', solve([]), None)
check('range-field-combination fixture 3', solve([["bytes",["0-1"]]]), ["bytes",["0-1"]])
check('range-field-combination fixture 4', solve([["items",["a"]],["items",["b"]]]), ["items",["a","b"]])
check('range-field-combination fixture 5', solve([["bytes",[]],["bytes",["0-0"]]]), ["bytes",["0-0"]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Repeated Range fields combine their member lists under one unit.

Parsed fields are [unit,members]. Return [unit,joined-members] when every unit matches first; mixed units return None; no fields returns None. Units already normalized.

split: open-access


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

N = 1
observations = []
def solve(members):
    return [m.strip() for m in members if m.strip()]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-member-empty-reject fixture 0', solve( ["0-"+str(N),""]), None)
check('range-member-empty-reject fixture 1', solve([" ","0-1"]), None)
check('range-member-empty-reject fixture 2', solve([]), None)
check('range-member-empty-reject fixture 3', solve([" 0-1 "]), ["0-1"])
check('range-member-empty-reject fixture 4', solve(["0-1","2-3"]), ["0-1","2-3"])
check('range-member-empty-reject fixture 5', solve(["-"+str(N)]), ["-"+str(N)])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(members):
    return [m.strip() for m in members] if members else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-member-empty-reject fixture 0', solve( ["0-"+str(N),""]), None)
check('range-member-empty-reject fixture 1', solve([" ","0-1"]), None)
check('range-member-empty-reject fixture 2', solve([]), None)
check('range-member-empty-reject fixture 3', solve([" 0-1 "]), ["0-1"])
check('range-member-empty-reject fixture 4', solve(["0-1","2-3"]), ["0-1","2-3"])
check('range-member-empty-reject fixture 5', solve(["-"+str(N)]), ["-"+str(N)])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
An empty list member invalidates this strict Range parser.

Already comma-split strings are trimmed; any empty member rejects the whole field with None. Otherwise return trimmed members, preserving order. Empty list rejects.

split: open-access


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

N = 1
observations = []
def solve(digits):
    return int(digits) if digits and digits.isdecimal() else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-decimal-ascii-only fixture 0', solve("١"*N), None)
check('range-decimal-ascii-only fixture 1', solve("１"), None)
check('range-decimal-ascii-only fixture 2', solve(str(N)), N)
check('range-decimal-ascii-only fixture 3', solve("000"), 0)
check('range-decimal-ascii-only fixture 4', solve(""), None)
check('range-decimal-ascii-only fixture 5', solve("-1"), 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(digits):
    return int(digits) if digits and all(c.isdigit() for c in digits) and all(c not in "²³¹" for c in digits) else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('range-decimal-ascii-only fixture 0', solve("١"*N), None)
check('range-decimal-ascii-only fixture 1', solve("１"), None)
check('range-decimal-ascii-only fixture 2', solve(str(N)), N)
check('range-decimal-ascii-only fixture 3', solve("000"), 0)
check('range-decimal-ascii-only fixture 4', solve(""), None)
check('range-decimal-ascii-only fixture 5', solve("-1"), 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:
Byte-range numbers reject non-ASCII decimal lookalikes.

Return integer for nonempty ASCII decimal digits only; otherwise None. This field parser does not accept signs or Unicode digit characters.

split: open-access