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

N = 1
observations = []
def solve(events):
    return sum(e['amount'] for e in events)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
events = [{'id': 'a', 'amount': N}, {'id': 'a', 'amount': N}, {'id': 'b', 'amount': N}]
check('replayed delivery', solve(events), 2*N)
check('empty stream', solve([]), 0)
check('distinct equal amounts', solve(events[1:]), 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)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

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

N = 1
observations = []
def solve(events):
    return sum(set(e['amount'] for e in events))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
events = [{'id': 'a', 'amount': N}, {'id': 'a', 'amount': N}, {'id': 'b', 'amount': N}]
check('replayed delivery', solve(events), 2*N)
check('empty stream', solve([]), 0)
check('distinct equal amounts', solve(events[1:]), 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:
A replayed event increases a total that should change once per event identity.

Deduplicate by event identity, preserving distinct events with equal amounts.

split: open-access


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

N = 1
observations = []
def solve(value, now, expires):
    return value if now <= expires else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('exact expiration', solve('record', N, N), None)
check('last valid half tick', solve('record', N-0.5, N), 'record')
check('past deadline', solve('record', N+1, N), 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(value, now, expires):
    return value if now < expires - 1 else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('exact expiration', solve('record', N, N), None)
check('last valid half tick', solve('record', N-0.5, N), 'record')
check('past deadline', solve('record', N+1, N), 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:
An entry remains visible at precisely its expiration instant.

Treat the valid interval as insertion time inclusive and expiration time exclusive.

split: open-access


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

N = 1
observations = []
def solve(rows, cursor):
    return [r for r in rows if r[0] > cursor[0]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
rows = [[N, 1], [N, 2], [N+1, 3]]
check('tied next record', solve(rows, [N, 1]), rows[1:])
check('last record', solve(rows, [N+1, 3]), [])
check('before first', solve(rows, [N-1, 0]), rows)
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(rows, cursor):
    return [r for r in rows if r[0] >= cursor[0]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
rows = [[N, 1], [N, 2], [N+1, 3]]
check('tied next record', solve(rows, [N, 1]), rows[1:])
check('last record', solve(rows, [N+1, 3]), [])
check('before first', solve(rows, [N-1, 0]), rows)
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 next page omits records sharing the final sort key of the previous page.

Compare the complete ordered key, including the unique record identifier.

split: open-access


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

N = 1
observations = []
def solve(base, deltas):
    return base + deltas[-1] if deltas else base
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('two independent updates', solve(N, [2, 2]), N+4)
check('mixed deltas', solve(N, [1, -1, 3]), N+3)
check('no updates', solve(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(base, deltas):
    return base + sum(set(deltas))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('two independent updates', solve(N, [2, 2]), N+4)
check('mixed deltas', solve(N, [1, -1, 3]), N+3)
check('no updates', solve(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:
Two updates calculated from the same snapshot leave only one increment.

Apply each delta to the current accumulator rather than replacing it from a stale base.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
from decimal import Decimal, ROUND_HALF_UP
N = 1
observations = []
def solve(value):
    return format(round(float(value), 2), '.2f')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('decimal midpoint', solve('1.005'), '1.01')
check('another midpoint', solve('2.675'), '2.68')
check('whole value', solve(str(N)), f'{N}.00')
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
from decimal import Decimal, ROUND_HALF_UP
N = 1
observations = []
def solve(value):
    return str(Decimal(float(value)).quantize(Decimal('0.01'), rounding=ROUND_HALF_UP))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('decimal midpoint', solve('1.005'), '1.01')
check('another midpoint', solve('2.675'), '2.68')
check('whole value', solve(str(N)), f'{N}.00')
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 value specified as a decimal rounds below the required half-up result.

Parse the decimal string directly and apply the explicit half-up rule.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import unicodedata
N = 1
observations = []
def solve(names):
    return len(set(names))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('composed variants', solve(['CAFÉ', 'Cafe\u0301', 'café'] * N), 1)
check('different names', solve(['north', 'south']), 2)
check('empty list', 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
import unicodedata
N = 1
observations = []
def solve(names):
    return len({unicodedata.normalize('NFC', name) for name in names})
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('composed variants', solve(['CAFÉ', 'Cafe\u0301', 'café'] * N), 1)
check('different names', solve(['north', 'south']), 2)
check('empty list', 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:
Case and Unicode composition produce multiple identities for equivalent names.

Normalize to NFC and case-fold each name before comparing identities.

split: open-access


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

N = 1
observations = []
def solve(batch, failed_index):
    return len(batch)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
batch = list(range(N+3))
check('partial failure', solve(batch, N), N)
check('first item fails', solve(batch, 0), 0)
check('complete batch', solve(batch, None), len(batch))
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(batch, failed_index):
    return max(0, failed_index - 1) if failed_index is not None else len(batch)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
batch = list(range(N+3))
check('partial failure', solve(batch, N), N)
check('first item fails', solve(batch, 0), 0)
check('complete batch', solve(batch, None), len(batch))
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:
After a partial batch failure, resumption skips work that never completed.

Persist the first unacknowledged index as the next resume position.

split: open-access


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

N = 1
observations = []
def solve(windows, t):
    return sum(start <= t <= end for start, end in windows)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
windows = [[0, N], [N, 2*N]]
check('shared boundary', solve(windows, N), 1)
check('first boundary', solve(windows, 0), 1)
check('outside final boundary', solve(windows, 2*N), 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(windows, t):
    return sum(start < t < end for start, end in windows)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
windows = [[0, N], [N, 2*N]]
check('shared boundary', solve(windows, N), 1)
check('first boundary', solve(windows, 0), 1)
check('outside final boundary', solve(windows, 2*N), 0)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A sample on a shared boundary is counted in both adjacent windows.

Use half-open intervals consistently across adjacent windows.

split: open-access


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

N = 1
observations = []
def solve(base, attempt, cap):
    return base * 2 ** attempt
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('bounded late retry', solve(N, 10, 30), 30)
check('initial retry', solve(N, 0, 30), N)
check('zero base', solve(0, 4, 30), 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(base, attempt, cap):
    return max(cap, base * 2 ** attempt)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('bounded late retry', solve(N, 10, 30), 30)
check('initial retry', solve(N, 0, 30), N)
check('zero base', solve(0, 4, 30), 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 failures produce a retry delay larger than the configured limit.

Apply the configured upper bound after computing exponential delay.

split: open-access


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

N = 1
observations = []
def solve(values):
    return sum(x for x in values if x is not None) / len(values) if values else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('missing and zero', solve([None, 0, 2*N]), N)
check('all missing', solve([None, None]), None)
check('no measurements', solve([]), 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(values):
    present = [x for x in values if x]
    return sum(present) / len(present) if present else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('missing and zero', solve([None, 0, 2*N]), N)
check('all missing', solve([None, None]), None)
check('no measurements', solve([]), 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:
Absent measurements change the mean even though no observation was recorded.

Use the same set of present observations for both numerator and denominator.

split: open-access


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

N = 1
observations = []
def solve(epoch, value, writes):
    for next_epoch, next_value in writes:
        epoch, value = next_epoch, next_value
    return [epoch, value]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
e = 10*N
check('late revoked writer', solve(e, 'current', [[e-1, 'stale']]), [e, 'current'])
check('same holder writes twice', solve(e, 'a', [[e, 'b'], [e, 'c']]), [e, 'c'])
check('new holder then old arrival', solve(e, 'a', [[e+1, 'new'], [e, 'old']]), [e+1, 'new'])
check('multiple handoffs', solve(e, 'a', [[e+2, 'b'], [e+1, 'c'], [e+3, 'd']]), [e+3, 'd'])
check('idle resource', solve(e, 'kept', []), [e, 'kept'])
check('initial epoch may be zero', solve(0, None, [[0, N]]), [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(epoch, value, writes):
    for next_epoch, next_value in writes:
        if next_epoch > epoch:
            epoch, value = next_epoch, next_value
    return [epoch, value]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
e = 10*N
check('late revoked writer', solve(e, 'current', [[e-1, 'stale']]), [e, 'current'])
check('same holder writes twice', solve(e, 'a', [[e, 'b'], [e, 'c']]), [e, 'c'])
check('new holder then old arrival', solve(e, 'a', [[e+1, 'new'], [e, 'old']]), [e+1, 'new'])
check('multiple handoffs', solve(e, 'a', [[e+2, 'b'], [e+1, 'c'], [e+3, 'd']]), [e+3, 'd'])
check('idle resource', solve(e, 'kept', []), [e, 'kept'])
check('initial epoch may be zero', solve(0, None, [[0, N]]), [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:
An old lease holder writes after a higher fencing epoch has already committed.

Process [epoch,value] writes in arrival order. Accept epochs >= the stored epoch and return [largest accepted epoch,last accepted value]. Tokens are nonnegative and uniquely allocated per lease.

split: open-access


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

N = 1
observations = []
def solve(lease, request, now, duration):
    return [lease[0], lease[1], now+duration] if request[0] == lease[0] else list(lease)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
lease = ['worker', N+1, 100+N]
check('previous incarnation renewal', solve(lease, ['worker', N], 90, 30), lease)
check('current incarnation renewal', solve(lease, ['worker', N+1], 90, N+20), ['worker', N+1, 110+N])
check('deadline already reached', solve(lease, ['worker', N+1], 100+N, 30), lease)
check('different owner', solve(lease, ['other', N+1], 90, 30), lease)
check('future generation is not ownership', solve(lease, ['worker', N+2], 90, 30), lease)
check('explicit zero extension', solve(lease, ['worker', N+1], 90, 0), ['worker', N+1, 90])
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(lease, request, now, duration):
    return [lease[0], lease[1], now+duration] if request[0] == lease[0] and now < lease[2] else list(lease)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
lease = ['worker', N+1, 100+N]
check('previous incarnation renewal', solve(lease, ['worker', N], 90, 30), lease)
check('current incarnation renewal', solve(lease, ['worker', N+1], 90, N+20), ['worker', N+1, 110+N])
check('deadline already reached', solve(lease, ['worker', N+1], 100+N, 30), lease)
check('different owner', solve(lease, ['other', N+1], 90, 30), lease)
check('future generation is not ownership', solve(lease, ['worker', N+2], 90, 30), lease)
check('explicit zero extension', solve(lease, ['worker', N+1], 90, 0), ['worker', N+1, 90])
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 delayed renewal is accepted after the same worker name has reacquired the lease.

A lease is [owner,generation,deadline]. A renewal names [owner,generation], now and nonnegative duration. If ownership matches and now < deadline, return a new deadline now+duration; otherwise return the unchanged lease. This models an atomic compare-and-swap at one authoritative clock.

split: open-access


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

N = 1
observations = []
def solve(left, right):
    a, b = sum(left.values()), sum(right.values())
    return 'equal' if a == b else ('before' if a < b else 'after')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('same-sum independent writes', solve({'a': N+1}, {'a': N, 'b': 1}), 'concurrent')
check('unequal-sum concurrent writes', solve({'a': N+2}, {'a': N, 'b': 1}), 'concurrent')
check('strict causal extension', solve({'a': N}, {'a': N, 'b': N}), 'before')
check('reverse causal extension', solve({'a': N+1, 'b': N}, {'a': N}), 'after')
check('missing zero component', solve({'a': N}, {'a': N, 'b': 0}), 'equal')
check('empty clocks', solve({}, {}), 'equal')
check('one replica progresses', solve({'a': N}, {'a': N+1}), 'before')
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(left, right):
    keys = sorted(set(left) | set(right))
    a = tuple(left.get(k, 0) for k in keys)
    b = tuple(right.get(k, 0) for k in keys)
    return 'equal' if a == b else ('before' if a < b else 'after')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('same-sum independent writes', solve({'a': N+1}, {'a': N, 'b': 1}), 'concurrent')
check('unequal-sum concurrent writes', solve({'a': N+2}, {'a': N, 'b': 1}), 'concurrent')
check('strict causal extension', solve({'a': N}, {'a': N, 'b': N}), 'before')
check('reverse causal extension', solve({'a': N+1, 'b': N}, {'a': N}), 'after')
check('missing zero component', solve({'a': N}, {'a': N, 'b': 0}), 'equal')
check('empty clocks', solve({}, {}), 'equal')
check('one replica progresses', solve({'a': N}, {'a': N+1}), 'before')
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 scalar or lexicographic comparison imposes an order on independent replica updates.

Given two maps from replica ID to nonnegative counter, return equal, before, after, or concurrent. Before requires every component <= and at least one <; after is its inverse.

split: open-access


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

N = 1
observations = []
def solve(members, votes):
    return bool(members) and len(votes) > len(set(members))//2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
members = list(range(2*N+1))
check('retransmitted single ballot', solve(members, [0]*(N+1)), False)
check('only foreign voters', solve(members, list(range(100, 101+N))), False)
check('exact eligible majority', solve(members, list(range(N+1))), True)
check('one below majority', solve(members, list(range(N))), False)
check('noise cannot complete minority', solve(members, list(range(N))+[999]), False)
check('repeats do not destroy real majority', solve(members, list(range(N+1))*N), True)
check('empty configuration', solve([], [0]), False)
check('single eligible member', solve(['only'], ['only']), True)
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, votes):
    return bool(members) and len(set(votes)) > len(set(members))//2
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
members = list(range(2*N+1))
check('retransmitted single ballot', solve(members, [0]*(N+1)), False)
check('only foreign voters', solve(members, list(range(100, 101+N))), False)
check('exact eligible majority', solve(members, list(range(N+1))), True)
check('one below majority', solve(members, list(range(N))), False)
check('noise cannot complete minority', solve(members, list(range(N))+[999]), False)
check('repeats do not destroy real majority', solve(members, list(range(N+1))*N), True)
check('empty configuration', solve([], [0]), False)
check('single eligible member', solve(['only'], ['only']), True)
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 coordinator declares a decision after counting retries or ballots from outside the membership.

Members define a fixed nonempty voting configuration. Return whether distinct eligible voter IDs number at least floor(member count/2)+1; an empty configuration cannot reach quorum.

split: open-access


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

N = 1
observations = []
def solve(floor, replicas):
    eligible = replicas
    return list(min(eligible, key=lambda r: (r[2], r[0]))[:2]) if eligible else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
v = N+10
check('stale fast replica', solve(v, [['a', v-1, 1], ['b', v, 10]]), ['b', v])
check('fast enough versus newer', solve(v, [['a', v, N], ['b', v+N, 20]]), ['a', v])
check('all replicas lag', solve(v, [['a', v-1, N]]), None)
check('exact floor accepted', solve(v, [['a', v, N]]), ['a', v])
check('stable latency tie', solve(v, [['b', v+1, N], ['a', v, N]]), ['a', v])
check('empty discovery result', solve(v, []), None)
check('zero-version session', solve(0, [['a', 0, N]]), ['a', 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(floor, replicas):
    eligible = [r for r in replicas if r[1] >= floor]
    return list(min(eligible, key=lambda r: (-r[1], r[2], r[0]))[:2]) if eligible else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
v = N+10
check('stale fast replica', solve(v, [['a', v-1, 1], ['b', v, 10]]), ['b', v])
check('fast enough versus newer', solve(v, [['a', v, N], ['b', v+N, 20]]), ['a', v])
check('all replicas lag', solve(v, [['a', v-1, N]]), None)
check('exact floor accepted', solve(v, [['a', v, N]]), ['a', v])
check('stable latency tie', solve(v, [['b', v+1, N], ['a', v, N]]), ['a', v])
check('empty discovery result', solve(v, []), None)
check('zero-version session', solve(0, [['a', 0, N]]), ['a', 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:
The fastest replica returns data older than a version this session has already seen.

Replicas are [ID,applied version,latency]. Return [ID,version] for the minimum (latency,ID) among versions >= session floor, or None if no replica qualifies. Version order is globally comparable in this model.

split: open-access


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

N = 1
observations = []
def solve(records):
    live = [r for r in records if r[1] == 'put']
    return max(live, key=lambda r: r[0])[2] if live else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('new delete suppresses stale put', solve([[N, 'put', 'old'], [N+1, 'delete', None]]), None)
check('later recreation', solve([[N, 'delete', None], [N+1, 'put', 'new']]), 'new')
check('delete wins version tie', solve([[N, 'put', 'old'], [N, 'delete', None]]), None)
check('tie independent of arrival', solve([[N, 'delete', None], [N, 'put', 'old']]), None)
check('newer of two live values', solve([[N, 'put', 'old'], [N+2, 'put', 'new']]), 'new')
check('empty replicas', solve([]), None)
check('only tombstone', solve([[N, 'delete', None]]), 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(records):
    if any(r[1] == 'delete' for r in records):
        return None
    return max(records, key=lambda r: r[0])[2] if records else None
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('new delete suppresses stale put', solve([[N, 'put', 'old'], [N+1, 'delete', None]]), None)
check('later recreation', solve([[N, 'delete', None], [N+1, 'put', 'new']]), 'new')
check('delete wins version tie', solve([[N, 'put', 'old'], [N, 'delete', None]]), None)
check('tie independent of arrival', solve([[N, 'delete', None], [N, 'put', 'old']]), None)
check('newer of two live values', solve([[N, 'put', 'old'], [N+2, 'put', 'new']]), 'new')
check('empty replicas', solve([]), None)
check('only tombstone', solve([[N, 'delete', None]]), None)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A surviving old value wins because the merge removes deletion markers before comparing versions.

Records are [version,kind,value] with kind put or delete. Return the latest value or None when the latest record is a deletion; deletion wins an equal-version put/delete tie. Equal-version put values are assumed identical; the model has no wall-clock conflict resolution.

split: open-access


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

N = 1
observations = []
def solve(total, demands):
    return [min(total, demand) for demand in demands]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('one shared allowance', solve(N, [N, N]), [(N+1)//2, N//2])
check('idle branch yields allowance', solve(N, [0, N]), [0, N])
check('leftover unit stays useful', solve(2*N+1, [N+1, N+1]), [N+1, N])
check('demand lower than allowance', solve(10*N, [N, 1]), [N, 1])
check('zero allowance', solve(0, [N, N]), [0, 0])
check('no branches', solve(N, []), [])
check('all branches idle', solve(N, [0, 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(total, demands):
    share = total//len(demands) if demands else 0
    return [min(share, demand) for demand in demands]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('one shared allowance', solve(N, [N, N]), [(N+1)//2, N//2])
check('idle branch yields allowance', solve(N, [0, N]), [0, N])
check('leftover unit stays useful', solve(2*N+1, [N+1, N+1]), [N+1, N])
check('demand lower than allowance', solve(10*N, [N, 1]), [N, 1])
check('zero allowance', solve(0, [N, N]), [0, 0])
check('no branches', solve(N, []), [])
check('all branches idle', solve(N, [0, 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:
Every child RPC spends the entire retry budget independently.

Given a nonnegative total allowance and nonnegative per-branch demand, return grants in branch order. Grant one unit per nonempty branch per round until total or all demand is exhausted. This models serialized admission to an already atomic shared counter.

split: open-access


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

N = 1
observations = []
def solve(events):
    active, deliveries, started = {}, [], 0
    for kind, key, value in events:
        if kind == 'start':
            started += 1
            active.setdefault(key, []).append(value)
        else:
            deliveries.extend([[waiter, value] for waiter in active.pop(key, [])])
    return [started, deliveries]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
starts = [['start', 'k', 'w'+str(i)] for i in range(N+1)]
check('overlapping callers share work', solve(starts+[['complete', 'k', N]]), [1, [['w'+str(i), N] for i in range(N+1)]])
check('later request starts fresh', solve([['start', 'k', 'a'], ['complete', 'k', N], ['start', 'k', 'b'], ['complete', 'k', N+1]]), [2, [['a', N], ['b', N+1]]])
check('separate keys remain separate', solve([['start', 'x', 'a'], ['start', 'y', 'b'], ['complete', 'y', N], ['complete', 'x', N+1]]), [2, [['b', N], ['a', N+1]]])
check('unsolicited completion', solve([['complete', 'unknown', N]]), [0, []])
check('empty arrival stream', solve([]), [0, []])
check('unfinished flight is pending', solve(starts), [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(events):
    active, completed, deliveries, started = {}, {}, [], 0
    for kind, key, value in events:
        if kind == 'start':
            if key in completed:
                deliveries.append([value, completed[key]])
            else:
                if key not in active:
                    started += 1
                active.setdefault(key, []).append(value)
        elif key in active:
            deliveries.extend([[waiter, value] for waiter in active.pop(key)])
            completed[key] = value
    return [started, deliveries]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
starts = [['start', 'k', 'w'+str(i)] for i in range(N+1)]
check('overlapping callers share work', solve(starts+[['complete', 'k', N]]), [1, [['w'+str(i), N] for i in range(N+1)]])
check('later request starts fresh', solve([['start', 'k', 'a'], ['complete', 'k', N], ['start', 'k', 'b'], ['complete', 'k', N+1]]), [2, [['a', N], ['b', N+1]]])
check('separate keys remain separate', solve([['start', 'x', 'a'], ['start', 'y', 'b'], ['complete', 'y', N], ['complete', 'x', N+1]]), [2, [['b', N], ['a', N+1]]])
check('unsolicited completion', solve([['complete', 'unknown', N]]), [0, []])
check('empty arrival stream', solve([]), [0, []])
check('unfinished flight is pending', solve(starts), [1, []])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Concurrent callers start separate fetches, or a completed fetch incorrectly absorbs a later request.

Events are ['start',key,waiter] or ['complete',key,value]. Distinct waiters share an active job. Completion emits [waiter,value] in registration order and removes that flight. Later starts create new jobs. Return [jobs started,deliveries]; unsolicited completions are ignored.

split: open-access


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

N = 1
observations = []
def solve(generation, message, acknowledgments):
    return any(ack_message == message for ack_generation, ack_message in acknowledgments)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
g, message = N+1, 'message-'+str(N)
check('prior owner late ack', solve(g, message, [[g-1, message]]), False)
check('wrong message same assignment', solve(g, message, [[g, 'different']]), False)
check('valid acknowledgment', solve(g, message, [[g, message]]), True)
check('future generation is not this delivery', solve(g, message, [[g+1, message]]), False)
check('valid among delayed noise', solve(g, message, [[g-1, message], [g, 'different'], [g, message]]), True)
check('no acknowledgment yet', solve(g, message, []), False)
check('invalid after valid cannot undo ack', solve(g, message, [[g, message], [g-1, message]]), True)
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(generation, message, acknowledgments):
    return any(ack_generation == generation for ack_generation, ack_message in acknowledgments)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
g, message = N+1, 'message-'+str(N)
check('prior owner late ack', solve(g, message, [[g-1, message]]), False)
check('wrong message same assignment', solve(g, message, [[g, 'different']]), False)
check('valid acknowledgment', solve(g, message, [[g, message]]), True)
check('future generation is not this delivery', solve(g, message, [[g+1, message]]), False)
check('valid among delayed noise', solve(g, message, [[g-1, message], [g, 'different'], [g, message]]), True)
check('no acknowledgment yet', solve(g, message, []), False)
check('invalid after valid cannot undo ack', solve(g, message, [[g, message], [g-1, message]]), True)
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 acknowledgment from a prior assignment is accepted after a rebalance reuses the message identifier.

The active delivery is [assignment generation,message ID]. Acks have the same shape. Return whether any acknowledgment exactly matches both fields. Other deliveries do not share this state.

split: open-access


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

N = 1
observations = []
def solve(state, local_log, requests):
    term, voted = state
    grants = []
    for request_term, candidate, last_term, last_index in requests:
        if request_term > term:
            term, voted = request_term, None
        allowed = request_term == term and last_index >= local_log[1] and voted in (None, candidate)
        if allowed:
            voted = candidate
        grants.append(allowed)
    return [term, voted, grants]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
t = N+10
check('longer old-term log rejected', solve([t, None], [N+2, N], [[t, 'a', N+1, 100*N]]), [t, None, [False]])
check('new term clears old vote', solve([t, 'old'], [N, N], [[t+1, 'new', N, N]]), [t+1, 'new', [True]])
check('newer log term beats length', solve([t, None], [N, 100*N], [[t, 'a', N+1, 1]]), [t, 'a', [True]])
check('one candidate per term', solve([t, None], [N, N], [[t, 'a', N, N], [t, 'b', N, N]]), [t, 'a', [True, False]])
check('same candidate retransmission', solve([t, 'a'], [N, N], [[t, 'a', N, N]]), [t, 'a', [True]])
check('old election term ignored', solve([t, None], [N, N], [[t-1, 'a', N+2, N]]), [t, None, [False]])
check('rejected new term still advances', solve([t, 'old'], [N+2, N], [[t+1, 'a', N, N]]), [t+1, None, [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(state, local_log, requests):
    term, voted = state
    grants = []
    for request_term, candidate, last_term, last_index in requests:
        term = max(term, request_term)
        allowed = request_term == term and (last_term, last_index) >= tuple(local_log) and voted in (None, candidate)
        if allowed:
            voted = candidate
        grants.append(allowed)
    return [term, voted, grants]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
t = N+10
check('longer old-term log rejected', solve([t, None], [N+2, N], [[t, 'a', N+1, 100*N]]), [t, None, [False]])
check('new term clears old vote', solve([t, 'old'], [N, N], [[t+1, 'new', N, N]]), [t+1, 'new', [True]])
check('newer log term beats length', solve([t, None], [N, 100*N], [[t, 'a', N+1, 1]]), [t, 'a', [True]])
check('one candidate per term', solve([t, None], [N, N], [[t, 'a', N, N], [t, 'b', N, N]]), [t, 'a', [True, False]])
check('same candidate retransmission', solve([t, 'a'], [N, N], [[t, 'a', N, N]]), [t, 'a', [True]])
check('old election term ignored', solve([t, None], [N, N], [[t-1, 'a', N+2, N]]), [t, None, [False]])
check('rejected new term still advances', solve([t, 'old'], [N+2, N], [[t+1, 'a', N, N]]), [t+1, None, [False]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A candidate with more entries from an older term receives a vote over the voter's newer history.

State is [current term,voted candidate or None], local log is [last term,last index], and requests are [election term,candidate,last log term,last index]. Newer terms clear the vote even if the candidate is rejected. Grant only in current term to an up-to-date log and an unclaimed or same vote. Return [term,voted candidate,grant decisions]. This is a voting-rule model, not a consensus implementation.

split: open-access


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

N = 1
observations = []
def solve(actions):
    pending, business, published = None, None, []
    for action in actions:
        if action[0] == 'stage':
            pending = action[1:]
            published.append(pending[0])
        elif action[0] == 'commit' and pending is not None:
            business = pending[1]
            pending = None
        elif action[0] == 'rollback':
            pending = None
    return [business, published]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('rollback cannot publish', solve([['stage', 'a', N], ['rollback'], ['dispatch']]), [None, []])
check('commit before publication', solve([['stage', 'a', N], ['dispatch']]), [None, []])
check('all committed events survive delay', solve([['stage', 'a', N], ['commit'], ['stage', 'b', N+1], ['commit'], ['dispatch']]), [N+1, ['a', 'b']])
check('single committed event', solve([['stage', 'a', N], ['commit'], ['dispatch']]), [N, ['a']])
check('repeat dispatch does not resend', solve([['stage', 'a', N], ['commit'], ['dispatch'], ['dispatch']]), [N, ['a']])
check('rollback preserves prior outbox', solve([['stage', 'a', N], ['commit'], ['stage', 'b', N+1], ['rollback'], ['dispatch']]), [N, ['a']])
check('empty transaction stream', solve([]), [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(actions):
    pending, business, latest, published = None, None, None, []
    for action in actions:
        if action[0] == 'stage':
            pending = action[1:]
        elif action[0] == 'commit' and pending is not None:
            business, latest = pending[1], pending[0]
            pending = None
        elif action[0] == 'rollback':
            pending = None
        elif action[0] == 'dispatch' and latest is not None:
            published.append(latest)
            latest = None
    return [business, published]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('rollback cannot publish', solve([['stage', 'a', N], ['rollback'], ['dispatch']]), [None, []])
check('commit before publication', solve([['stage', 'a', N], ['dispatch']]), [None, []])
check('all committed events survive delay', solve([['stage', 'a', N], ['commit'], ['stage', 'b', N+1], ['commit'], ['dispatch']]), [N+1, ['a', 'b']])
check('single committed event', solve([['stage', 'a', N], ['commit'], ['dispatch']]), [N, ['a']])
check('repeat dispatch does not resend', solve([['stage', 'a', N], ['commit'], ['dispatch'], ['dispatch']]), [N, ['a']])
check('rollback preserves prior outbox', solve([['stage', 'a', N], ['commit'], ['stage', 'b', N+1], ['rollback'], ['dispatch']]), [N, ['a']])
check('empty transaction stream', solve([]), [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:
An event escapes before its associated business state commits, or committed events disappear before dispatch.

Actions are ['stage',ID,value], ['commit'], ['rollback'], or ['dispatch']. Stage replaces an uncommitted pending write; commit atomically stores its value and appends its event; rollback discards pending work. Dispatch publishes each locally pending committed event once. IDs are unique per committed transaction. Return [business value,published event IDs]. No transport crash or exactly-once external-delivery guarantee is modeled.

split: open-access


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

N = 1
observations = []
def solve(initial, deliveries):
    applied = set(initial)
    for identity, dependencies in deliveries:
        if set(dependencies) <= applied:
            applied.add(identity)
    return [sorted(applied), []]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
ids = ['op-'+str(i) for i in range(N+3)]
chain = [[identity, [ids[i-1]] if i else []] for i, identity in enumerate(ids)]
check('reverse dependency chain', solve([], list(reversed(chain))), [sorted(ids), []])
check('already causal arrival', solve([], chain), [sorted(ids), []])
check('missing prerequisite retained', solve([], [['waiting', ['absent']]]), [[], ['waiting']])
check('initial context unlocks delivery', solve(['root'], [['child', ['root']]]), [['child', 'root'], []])
check('dependency cycle stays pending', solve([], [['a', ['b']], ['b', ['a']]]), [[], ['a', 'b']])
check('replayed delivery', solve([], chain+chain), [sorted(ids), []])
check('empty delivery', solve(['root'], []), [['root'], []])
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(initial, deliveries):
    applied = set(initial)
    pending = {identity: set(dependencies) for identity, dependencies in deliveries if identity not in applied}
    for identity, dependencies in list(pending.items()):
        if dependencies <= applied:
            applied.add(identity)
            del pending[identity]
    return [sorted(applied), sorted(pending)]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
ids = ['op-'+str(i) for i in range(N+3)]
chain = [[identity, [ids[i-1]] if i else []] for i, identity in enumerate(ids)]
check('reverse dependency chain', solve([], list(reversed(chain))), [sorted(ids), []])
check('already causal arrival', solve([], chain), [sorted(ids), []])
check('missing prerequisite retained', solve([], [['waiting', ['absent']]]), [[], ['waiting']])
check('initial context unlocks delivery', solve(['root'], [['child', ['root']]]), [['child', 'root'], []])
check('dependency cycle stays pending', solve([], [['a', ['b']], ['b', ['a']]]), [[], ['a', 'b']])
check('replayed delivery', solve([], chain+chain), [sorted(ids), []])
check('empty delivery', solve(['root'], []), [['root'], []])
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 message whose prerequisites arrive later is dropped or never revisited after another pending message becomes ready.

Initial IDs are already applied. Deliveries are [ID,dependency IDs], with duplicate IDs denoting the same operation. Apply an ID only after all dependencies are applied. Return sorted applied and pending IDs after exhausting all currently satisfiable dependencies; missing dependencies and cycles remain pending. Payload conflict resolution is outside this model.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import sqlite3
N = 1
observations = []
def solve(candidates, excluded):
    db = sqlite3.connect(':memory:')
    try:
        db.execute('CREATE TABLE candidates (v INTEGER)')
        db.execute('CREATE TABLE excluded (v INTEGER)')
        db.executemany('INSERT INTO candidates VALUES (?)', [(v,) for v in candidates])
        db.executemany('INSERT INTO excluded VALUES (?)', [(v,) for v in excluded])
        return [row[0] for row in db.execute('SELECT c.v FROM candidates AS c WHERE c.v NOT IN (SELECT v FROM excluded) ORDER BY c.rowid')]
    finally:
        db.close()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unknown exclusion does not poison other values', solve([N, N+1], [None, N]), [N+1])
check('unknown candidate is not equal to a known exclusion', solve([None, N+1], [N]), [None, N+1])
check('NULL is not equal to NULL', solve([None, N], [None]), [None, N])
check('empty exclusion retains every occurrence', solve([N, N, None], []), [N, N, None])
check('exact known matches are excluded', solve([N, N+1, N+2], [N+2, N]), [N+1])
check('left multiplicity survives unrelated right duplicates', solve([N, N, N+1], [N+1, N+1]), [N, N])
check('empty candidates', solve([], [None, 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)

evaluation_group: xs-anti-join

source:
"""Failure Map reference implementation. Python standard library only."""
import json
import sqlite3
N = 1
observations = []
def solve(candidates, excluded):
    db = sqlite3.connect(':memory:')
    try:
        db.execute('CREATE TABLE candidates (v INTEGER)')
        db.execute('CREATE TABLE excluded (v INTEGER)')
        db.executemany('INSERT INTO candidates VALUES (?)', [(v,) for v in candidates])
        db.executemany('INSERT INTO excluded VALUES (?)', [(v,) for v in excluded])
        return [row[0] for row in db.execute('SELECT c.v FROM candidates AS c WHERE c.v NOT IN (SELECT v FROM excluded WHERE v IS NOT NULL) ORDER BY c.rowid')]
    finally:
        db.close()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unknown exclusion does not poison other values', solve([N, N+1], [None, N]), [N+1])
check('unknown candidate is not equal to a known exclusion', solve([None, N+1], [N]), [None, N+1])
check('NULL is not equal to NULL', solve([None, N], [None]), [None, N])
check('empty exclusion retains every occurrence', solve([N, N, None], []), [N, N, None])
check('exact known matches are excluded', solve([N, N+1, N+2], [N+2, N]), [N+1])
check('left multiplicity survives unrelated right duplicates', solve([N, N, N+1], [N+1, N+1]), [N, N])
check('empty candidates', solve([], [None, 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 single unknown exclusion value removes unrelated candidates from an anti-join result.

For each candidate in input order, retain it unless an excluded non-NULL value is SQL-equal to it. NULL never equals any value, including NULL. Duplicates on the left remain duplicates. Inputs are integers or None.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import sqlite3
N = 1
observations = []
def solve(values):
    db = sqlite3.connect(':memory:')
    try:
        db.execute('CREATE TABLE measurements (v INTEGER)')
        db.executemany('INSERT INTO measurements VALUES (?)', [(v,) for v in values])
        return list(db.execute('SELECT COALESCE(SUM(v), 0), COUNT(v), COUNT(*) FROM measurements').fetchone())
    finally:
        db.close()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('no input rows', solve([]), [None, 0, 0])
check('all rows unknown', solve([None] * N), [None, 0, N])
check('a real zero is a measurement', solve([0]), [0, 1, 1])
check('positive and negative values cancel', solve([N, -N, None]), [0, 2, 3])
check('known values amid unknowns', solve([N, None, 2*N]), [3*N, 2, 3])
check('negative aggregate is retained', solve([-N, -N-1]), [-2*N-1, 2, 2])
check('repeated zero rows retain counts', solve([0] * (N+1)), [0, N+1, N+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
import sqlite3
N = 1
observations = []
def solve(values):
    db = sqlite3.connect(':memory:')
    try:
        db.execute('CREATE TABLE measurements (v INTEGER)')
        db.executemany('INSERT INTO measurements VALUES (?)', [(v,) for v in values])
        return list(db.execute('SELECT NULLIF(SUM(v), 0), COUNT(v), COUNT(*) FROM measurements').fetchone())
    finally:
        db.close()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('no input rows', solve([]), [None, 0, 0])
check('all rows unknown', solve([None] * N), [None, 0, N])
check('a real zero is a measurement', solve([0]), [0, 1, 1])
check('positive and negative values cancel', solve([N, -N, None]), [0, 2, 3])
check('known values amid unknowns', solve([N, None, 2*N]), [3*N, 2, 3])
check('negative aggregate is retained', solve([-N, -N-1]), [-2*N-1, 2, 2])
check('repeated zero rows retain counts', solve([0] * (N+1)), [0, N+1, N+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:
An empty or all-NULL aggregate is rendered as zero, while an attempted repair also erases genuine zero totals.

Return [sum,count_of_non_NULL_values,total_row_count]. Sum is None when there are no non-NULL values, otherwise their exact integer sum, including zero. A scalar aggregate always returns this one triple.

split: open-access


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

N = 1
observations = []
def solve(lines, tags):
    return [sum(amount for amount in lines for tag in tags), sorted(set(tags))]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('two lines crossed with multiple tags', solve([N, N+1], ['red', 'blue']), [2*N+1, ['blue', 'red']])
check('equal-priced lines are distinct facts', solve([N] * (N+1), ['one']), [N*(N+1), ['one']])
check('untagged invoice remains in the result', solve([N, N+2], []), [2*N+2, []])
check('duplicate tag rows do not multiply revenue', solve([N], ['a', 'a', 'b']), [N, ['a', 'b']])
check('empty invoice preserves descriptive tags', solve([], ['b', 'a']), [0, ['a', 'b']])
check('credit and charge lines remain separate', solve([N, -N, N], ['credit']), [N, ['credit']])
check('no lines or tags', 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(lines, tags):
    return [sum(set(lines)), sorted(set(tags))]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('two lines crossed with multiple tags', solve([N, N+1], ['red', 'blue']), [2*N+1, ['blue', 'red']])
check('equal-priced lines are distinct facts', solve([N] * (N+1), ['one']), [N*(N+1), ['one']])
check('untagged invoice remains in the result', solve([N, N+2], []), [2*N+2, []])
check('duplicate tag rows do not multiply revenue', solve([N], ['a', 'a', 'b']), [N, ['a', 'b']])
check('empty invoice preserves descriptive tags', solve([], ['b', 'a']), [0, ['a', 'b']])
check('credit and charge lines remain separate', solve([N, -N, N], ['credit']), [N, ['credit']])
check('no lines or tags', 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:
Adding multiple descriptive tags inflates the subtotal by repeating each invoice line in a join product.

Given one invoice's integer line amounts and descriptive tag strings, return [sum_of_all_lines, sorted_unique_tags]. Each line occurrence contributes once, even equal-priced lines. Missing tags do not remove an invoice or its revenue.

split: open-access


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

N = 1
observations = []
def solve(versions, cutoff):
    chosen = {}
    for key, seq, value in versions:
        if key not in chosen or seq > chosen[key][0]:
            chosen[key] = (seq, value)
    return {key: pair[1] for key, pair in chosen.items()}
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
base = 10*N
check('future committed version is invisible', solve([['a', base, 'seen'], ['a', base+1, 'future']], base), {'a': 'seen'})
check('old physical row does not win', solve([['a', base-1, 'old'], ['a', base, 'new']], base), {'a': 'new'})
check('descending storage order is also valid', solve([['a', base+2, 'later'], ['a', base, N], ['a', base-2, 'old']], base), {'a': N})
check('each key has its own newest visible version', solve([['a', base, 1], ['b', base-1, 2], ['a', base+1, 3]], base), {'a': 1, 'b': 2})
check('key first created after cutoff is absent', solve([['a', base+1, N]], base), {})
check('nullable value remains a visible row', solve([['a', base, None]], base), {'a': None})
check('empty version history', solve([], base), {})
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(versions, cutoff):
    chosen = {}
    for key, seq, value in versions:
        if seq <= cutoff and key not in chosen:
            chosen[key] = value
    return chosen
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
base = 10*N
check('future committed version is invisible', solve([['a', base, 'seen'], ['a', base+1, 'future']], base), {'a': 'seen'})
check('old physical row does not win', solve([['a', base-1, 'old'], ['a', base, 'new']], base), {'a': 'new'})
check('descending storage order is also valid', solve([['a', base+2, 'later'], ['a', base, N], ['a', base-2, 'old']], base), {'a': N})
check('each key has its own newest visible version', solve([['a', base, 1], ['b', base-1, 2], ['a', base+1, 3]], base), {'a': 1, 'b': 2})
check('key first created after cutoff is absent', solve([['a', base+1, N]], base), {})
check('nullable value remains a visible row', solve([['a', base, None]], base), {'a': None})
check('empty version history', solve([], base), {})
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 read returns the newest physical version even though that version was not visible at the reader's snapshot.

Versions are [key,commit_sequence,value], with integer commit sequences unique within a key. Return a map from key to the value at its greatest sequence <= cutoff; omit keys with no visible version. Input storage order is arbitrary.

split: open-access


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

N = 1
observations = []
def solve(rows):
    return len(set(map(tuple, rows))) == len(rows)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unknown second component can repeat', solve([[N, None], [N, None]]), True)
check('unknown first component can repeat', solve([[None, N], [None, N]]), True)
check('both components unknown can repeat', solve([[None, None]] * (N+1)), True)
check('fully known repeated key conflicts', solve([[N, N+1], [N, N+1]]), False)
check('known zero is not unknown', solve([[N, 0], [N, 0], [None, 0]]), False)
check('different complete tuples coexist', solve([[N, N+1], [N, N+2], [N+1, N+1]]), True)
check('empty relation satisfies uniqueness', solve([]), True)
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)

evaluation_group: xs-composite-index-conflicts

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

N = 1
observations = []
def solve(rows):
    comparable = [tuple(row) for row in rows if row[0] is not None]
    return len(set(comparable)) == len(comparable)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unknown second component can repeat', solve([[N, None], [N, None]]), True)
check('unknown first component can repeat', solve([[None, N], [None, N]]), True)
check('both components unknown can repeat', solve([[None, None]] * (N+1)), True)
check('fully known repeated key conflicts', solve([[N, N+1], [N, N+1]]), False)
check('known zero is not unknown', solve([[N, 0], [N, 0], [None, 0]]), False)
check('different complete tuples coexist', solve([[N, N+1], [N, N+2], [N+1, N+1]]), True)
check('empty relation satisfies uniqueness', solve([]), True)
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 tuples containing an unknown component are rejected as duplicate composite keys.

Return whether a list of two-component keys satisfies ordinary NULL-distinct SQL UNIQUE semantics: two rows conflict only when both components in both rows are non-NULL and equal. None represents SQL NULL.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
from bisect import bisect_left, bisect_right
N = 1
observations = []
def solve(rows, tenant):
    index = [tuple(row) for row in rows]
    return [list(row) for row in index[bisect_left(index, (tenant,)):bisect_right(index, (tenant,))]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
rows = [[N, -N, 'negative'], [N, 0, 'zero'], [N, N, 'positive'], [N+1, -N, 'next']]
check('entire composite prefix', solve(rows, N), rows[:3])
check('next tenant negative key is not leaked backward', solve([[N+1, -N, 'next']], N), [])
check('negative secondary key is not omitted', solve([[N, -N, 'kept']], N), [[N, -N, 'kept']])
check('negative leading key', solve([[-N, -2, 'a'], [-N, 1, 'b'], [0, 0, 'c']], -N), [[-N, -2, 'a'], [-N, 1, 'b']])
check('missing leading prefix', solve(rows, N+3), [])
check('positive-only prefix still works', solve([[N, 1, 'a'], [N, N+2, 'b']], N), [[N, 1, 'a'], [N, N+2, 'b']])
check('empty index', solve([], 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
from bisect import bisect_left, bisect_right
N = 1
observations = []
def solve(rows, tenant):
    index = [tuple(row) for row in rows]
    return [list(row) for row in index[bisect_left(index, (tenant, 0)):bisect_left(index, (tenant+1, 0))]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
rows = [[N, -N, 'negative'], [N, 0, 'zero'], [N, N, 'positive'], [N+1, -N, 'next']]
check('entire composite prefix', solve(rows, N), rows[:3])
check('next tenant negative key is not leaked backward', solve([[N+1, -N, 'next']], N), [])
check('negative secondary key is not omitted', solve([[N, -N, 'kept']], N), [[N, -N, 'kept']])
check('negative leading key', solve([[-N, -2, 'a'], [-N, 1, 'b'], [0, 0, 'c']], -N), [[-N, -2, 'a'], [-N, 1, 'b']])
check('missing leading prefix', solve(rows, N+3), [])
check('positive-only prefix still works', solve([[N, 1, 'a'], [N, N+2, 'b']], N), [[N, 1, 'a'], [N, N+2, 'b']])
check('empty index', solve([], 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 composite index lookup omits some rows for its leading key or leaks rows from the next leading key.

Input rows are sorted unique tuples [integer_tenant,integer_secondary,payload_string]. Return all rows for the requested tenant in index order. Secondary keys can be any integer. The model assumes integer, not string, leading keys.

split: open-access


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

N = 1
observations = []
def solve(columns, values):
    return {column[1]: value for column, value in zip(columns, values)}
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('joined identifiers both survive', solve([['orders', 'id'], ['users', 'id']], [N, N+1]), {'orders.id': N, 'users.id': N+1})
check('reversing projection preserves qualified schema', solve([['users', 'id'], ['orders', 'id']], [N+1, N]), {'orders.id': N, 'users.id': N+1})
check('nullable outer-join field is retained', solve([['left', 'id'], ['right', 'id']], [N, None]), {'left.id': N, 'right.id': None})
check('zero identifier is retained', solve([['left', 'id'], ['right', 'id']], [0, N]), {'left.id': 0, 'right.id': N})
check('ordinary names are qualified consistently', solve([['a', 'name'], ['b', 'amount']], ['case-'+str(N), N*3]), {'a.name': 'case-'+str(N), 'b.amount': N*3})
check('ordinal-looking source name cannot collide', solve([['a', 'id'], ['b', 'id'], ['c', 'id_2']], [N, N+1, N+2]), {'a.id': N, 'b.id': N+1, 'c.id_2': N+2})
check('empty projection', 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(columns, values):
    result = {}
    for column, value in zip(columns, values):
        name = column[1]
        suffix = 2
        while name in result:
            name = column[1] + '_' + str(suffix)
            suffix += 1
        result[name] = value
    return result
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('joined identifiers both survive', solve([['orders', 'id'], ['users', 'id']], [N, N+1]), {'orders.id': N, 'users.id': N+1})
check('reversing projection preserves qualified schema', solve([['users', 'id'], ['orders', 'id']], [N+1, N]), {'orders.id': N, 'users.id': N+1})
check('nullable outer-join field is retained', solve([['left', 'id'], ['right', 'id']], [N, None]), {'left.id': N, 'right.id': None})
check('zero identifier is retained', solve([['left', 'id'], ['right', 'id']], [0, N]), {'left.id': 0, 'right.id': N})
check('ordinary names are qualified consistently', solve([['a', 'name'], ['b', 'amount']], ['case-'+str(N), N*3]), {'a.name': 'case-'+str(N), 'b.amount': N*3})
check('ordinal-looking source name cannot collide', solve([['a', 'id'], ['b', 'id'], ['c', 'id_2']], [N, N+1, N+2]), {'a.id': N, 'b.id': N+1, 'c.id_2': N+2})
check('empty projection', 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:
Converting a joined result into an object overwrites one table's identifier with another table's identifier.

Each projection entry is [relation_alias,column_name] and has a matching value. Pairs are unique; components contain no dots. Return a map keyed by relation_alias + '.' + column_name, retaining every value including NULL and zero.

split: open-access


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

N = 1
observations = []
def solve(rows, threshold):
    totals = {}
    for group, amount in rows:
        if amount >= threshold:
            totals[group] = totals.get(group, 0) + amount
    return totals
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('small contributions jointly qualify', solve([['a', N], ['a', N]], 2*N), {'a': 2*N})
check('exact single contribution qualifies', solve([['a', 3*N]], 3*N), {'a': 3*N})
check('negative adjustment changes eligibility', solve([['a', 4*N], ['a', -3*N]], 2*N), {})
check('group-specific totals do not mingle', solve([['a', N], ['b', 5*N], ['a', 2*N]], 3*N), {'a': 3*N, 'b': 5*N})
check('below-threshold completed group', solve([['a', N], ['a', -1]], N), {})
check('zero threshold retains a real cancelling group', solve([['a', N], ['a', -N]], 0), {'a': 0})
check('empty relation does not invent a zero group', 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)

evaluation_group: xs-group-total-filter

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

N = 1
observations = []
def solve(rows, threshold):
    totals = {}
    for group, amount in rows:
        totals[group] = totals.get(group, 0) + amount
    return {group: total for group, total in totals.items() if total > threshold}
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('small contributions jointly qualify', solve([['a', N], ['a', N]], 2*N), {'a': 2*N})
check('exact single contribution qualifies', solve([['a', 3*N]], 3*N), {'a': 3*N})
check('negative adjustment changes eligibility', solve([['a', 4*N], ['a', -3*N]], 2*N), {})
check('group-specific totals do not mingle', solve([['a', N], ['b', 5*N], ['a', 2*N]], 3*N), {'a': 3*N, 'b': 5*N})
check('below-threshold completed group', solve([['a', N], ['a', -1]], N), {})
check('zero threshold retains a real cancelling group', solve([['a', N], ['a', -N]], 0), {'a': 0})
check('empty relation does not invent a zero group', 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:
A group made of individually small contributions disappears even though its total meets the reporting threshold.

Rows are [group_string,signed_integer_amount]. Return a map of existing groups to their full sum, keeping groups whose sum is >= threshold. Never invent groups for empty input; negative contributions must participate before filtering.

split: open-access


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

N = 1
observations = []
def solve(rows, k):
    ordered = sorted(rows, key=lambda row: (-row[1], row[0]))
    return ordered[:k]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('tie straddles a row-count boundary', solve([[3, N], [2, 3*N], [1, 3*N]], 1), [[1, 3*N], [2, 3*N]])
check('earlier tie consumes two row positions', solve([[4, N], [3, 2*N], [2, 3*N], [1, 3*N]], 2), [[1, 3*N], [2, 3*N]])
check('tie at a later cutoff includes every equal score', solve([[1, 4*N], [4, N], [3, 2*N], [2, 2*N]], 2), [[1, 4*N], [2, 2*N], [3, 2*N]])
check('zero limit', solve([[1, N]], 0), [])
check('oversized limit sorts all rows', solve([[2, -N], [1, 0]], N+3), [[1, 0], [2, -N]])
check('negative scores also tie', solve([[2, -N], [1, -N], [3, -N-1]], 1), [[1, -N], [2, -N]])
check('empty relation', solve([], 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(rows, k):
    levels = sorted({row[1] for row in rows}, reverse=True)[:k]
    return sorted([row for row in rows if row[1] in levels], key=lambda row: (-row[1], row[0]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('tie straddles a row-count boundary', solve([[3, N], [2, 3*N], [1, 3*N]], 1), [[1, 3*N], [2, 3*N]])
check('earlier tie consumes two row positions', solve([[4, N], [3, 2*N], [2, 3*N], [1, 3*N]], 2), [[1, 3*N], [2, 3*N]])
check('tie at a later cutoff includes every equal score', solve([[1, 4*N], [4, N], [3, 2*N], [2, 2*N]], 2), [[1, 4*N], [2, 2*N], [3, 2*N]])
check('zero limit', solve([[1, N]], 0), [])
check('oversized limit sorts all rows', solve([[2, -N], [1, 0]], N+3), [[1, 0], [2, -N]])
check('negative scores also tie', solve([[2, -N], [1, -N], [3, -N-1]], 1), [[1, -N], [2, -N]])
check('empty relation', solve([], 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:
Tie handling either truncates equivalent rows or admits an extra lower-scoring group beyond the kth-row boundary.

Rows are [unique_integer_id,integer_score]. For nonnegative k, return all rows tied with or above the kth row under score-descending ordering, with id ascending within ties. k=0 returns empty; k>=row_count returns all rows.

split: open-access


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

N = 1
observations = []
def solve(rows):
    groups = {}
    for first, second, amount in rows:
        key = first + second
        if key not in groups:
            groups[key] = [first, second, 0]
        groups[key][2] += amount
    return sorted(groups.values())
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unseparated component boundaries collide', solve([['ab', 'c', N], ['a', 'bc', N+1]]), [['a', 'bc', N+1], ['ab', 'c', N]])
check('delimiter occurs inside either component', solve([['a|b', 'c', N], ['a', 'b|c', N+1]]), [['a', 'b|c', N+1], ['a|b', 'c', N]])
check('empty components are ordered, not interchangeable', solve([['', 'x', N], ['x', '', N+1]]), [['', 'x', N], ['x', '', N+1]])
check('delimiter-only boundaries remain distinct', solve([['|', '', N], ['', '|', 2*N]]), [['', '|', 2*N], ['|', '', N]])
check('identical composite keys aggregate', solve([['team', str(N), N], ['team', str(N), N+2]]), [['team', str(N), 2*N+2]])
check('signed contributions stay within their group', solve([['a', 'b', N], ['a', 'b', -N], ['a', 'c', 2*N]]), [['a', 'b', 0], ['a', 'c', 2*N]])
check('empty relation', 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(rows):
    groups = {}
    for first, second, amount in rows:
        key = first + '|' + second
        if key not in groups:
            groups[key] = [first, second, 0]
        groups[key][2] += amount
    return sorted(groups.values())
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unseparated component boundaries collide', solve([['ab', 'c', N], ['a', 'bc', N+1]]), [['a', 'bc', N+1], ['ab', 'c', N]])
check('delimiter occurs inside either component', solve([['a|b', 'c', N], ['a', 'b|c', N+1]]), [['a', 'b|c', N+1], ['a|b', 'c', N]])
check('empty components are ordered, not interchangeable', solve([['', 'x', N], ['x', '', N+1]]), [['', 'x', N], ['x', '', N+1]])
check('delimiter-only boundaries remain distinct', solve([['|', '', N], ['', '|', 2*N]]), [['', '|', 2*N], ['|', '', N]])
check('identical composite keys aggregate', solve([['team', str(N), N], ['team', str(N), N+2]]), [['team', str(N), 2*N+2]])
check('signed contributions stay within their group', solve([['a', 'b', N], ['a', 'b', -N], ['a', 'c', 2*N]]), [['a', 'b', 0], ['a', 'c', 2*N]])
check('empty relation', 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:
Distinct two-column groups collapse together because their concatenated or delimiter-separated key strings coincide.

Rows contain [first_string,second_string,integer_amount]. Group by exact ordered pairs and return sorted [first,second,total] rows. Components may be empty or contain delimiter characters; input occurrences are all counted.

split: open-access


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

N = 1
observations = []
def solve(left, right):
    return [[a[0], b[0]] for a in left for b in right if a[1] <= b[1] < a[2]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('left interval is fully contained', solve([['a', N, 2*N]], [['b', 0, 3*N]]), [['a', 'b']])
check('right interval is fully contained', solve([['a', 0, 3*N]], [['b', N, 2*N]]), [['a', 'b']])
check('touching boundary is not an overlap', solve([['a', 0, N]], [['b', N, 2*N]]), [])
check('empty interval does not join an enclosing interval', solve([['a', N, N]], [['b', 0, 2*N]]), [])
check('right empty interval is not a point event', solve([['a', 0, 2*N]], [['b', N, N]]), [])
check('partial intersection crosses starting order', solve([['a', N, 3*N]], [['b', 0, 2*N]]), [['a', 'b']])
check('ordered many-to-many pairs', solve([['a', 0, 2*N], ['c', 2*N, 4*N]], [['b', N, 3*N], ['d', 4*N, 5*N]]), [['a', 'b'], ['c', 'b']])
check('empty relation', solve([], [['b', 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(left, right):
    return [[a[0], b[0]] for a in left for b in right if a[1] <= b[2] and b[1] <= a[2]]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('left interval is fully contained', solve([['a', N, 2*N]], [['b', 0, 3*N]]), [['a', 'b']])
check('right interval is fully contained', solve([['a', 0, 3*N]], [['b', N, 2*N]]), [['a', 'b']])
check('touching boundary is not an overlap', solve([['a', 0, N]], [['b', N, 2*N]]), [])
check('empty interval does not join an enclosing interval', solve([['a', N, N]], [['b', 0, 2*N]]), [])
check('right empty interval is not a point event', solve([['a', 0, 2*N]], [['b', N, N]]), [])
check('partial intersection crosses starting order', solve([['a', N, 3*N]], [['b', 0, 2*N]]), [['a', 'b']])
check('ordered many-to-many pairs', solve([['a', 0, 2*N], ['c', 2*N, 4*N]], [['b', N, 3*N], ['d', 4*N, 5*N]]), [['a', 'b'], ['c', 'b']])
check('empty relation', solve([], [['b', 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:
An interval join drops a left interval entirely enclosed by a right interval, then an endpoint-based repair creates false matches.

Each input row is [string_id,integer_start,integer_end] with start <= end. Treat intervals as half-open. Return [left_id,right_id] for every pair with a nonempty intersection, in left input order then right input order.

split: open-access


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

N = 1
observations = []
def solve(total, count, changes):
    for old, new in changes:
        if new is not None:
            total += new
            count += 1
    return [total, count]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('replacement changes sum but not row count', solve(N, 1, [[N, 3*N]]), [3*N, 1])
check('deleting the last row', solve(N, 1, [[N, None]]), [0, 0])
check('insertion creates a row', solve(0, 0, [[None, N]]), [N, 1])
check('zero is a present row value', solve(0, 1, [[0, N], [N, 0]]), [0, 1])
check('multiple replacements of the same row', solve(N, 1, [[N, N+1], [N+1, -N]]), [-N, 1])
check('insert and delete a zero row', solve(N, 1, [[None, 0], [0, None]]), [N, 1])
check('no changes preserve the original view', solve(2*N, 2, []), [2*N, 2])
check('different row operations compose', solve(3*N, 2, [[N, None], [2*N, 4*N], [None, -N]]), [3*N, 2])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(total, count, changes):
    for old, new in changes:
        total += (new if new is not None else 0) - (old if old is not None else 0)
        if new is not None:
            count += 1
        elif old is not None:
            count -= 1
    return [total, count]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('replacement changes sum but not row count', solve(N, 1, [[N, 3*N]]), [3*N, 1])
check('deleting the last row', solve(N, 1, [[N, None]]), [0, 0])
check('insertion creates a row', solve(0, 0, [[None, N]]), [N, 1])
check('zero is a present row value', solve(0, 1, [[0, N], [N, 0]]), [0, 1])
check('multiple replacements of the same row', solve(N, 1, [[N, N+1], [N+1, -N]]), [-N, 1])
check('insert and delete a zero row', solve(N, 1, [[None, 0], [0, None]]), [N, 1])
check('no changes preserve the original view', solve(2*N, 2, []), [2*N, 2])
check('different row operations compose', solve(3*N, 2, [[N, None], [2*N, 4*N], [None, -N]]), [3*N, 2])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
A materialized total accumulates superseded row values, while an attempted repair also counts updates as new rows.

Given a valid starting [sum,row_count] and an ordered list of [old_value,new_value] changes, return the updated pair. None means the row is absent, not a nullable stored value. Changes are valid and applied exactly once; no deduplication is requested.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import re
N = 1
observations = []
def solve(text):
    return text.split('|')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
label = 'field' * N
check('literal escaped separator', solve(label + r'\|inside|tail'), [label + '|inside', 'tail'])
check('even escape run leaves separator active', solve(label + '\\\\|tail'), [label + '\\', 'tail'])
check('multiple escaped backslashes', solve('\\\\' * N + '|end'), ['\\' * N, 'end'])
check('empty fields remain positioned', solve('|a||'), ['', 'a', '', ''])
check('empty record is one empty field', solve(''), [''])
check('dangling escape rejected', solve(label + '\\'), None)
check('unknown escape rejected', solve(label + r'\q'), 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
import re
N = 1
observations = []
def solve(text):
    fields = re.split(r'(?<!\\)\|', text)
    return [value.replace(r'\|', '|').replace('\\\\', '\\') for value in fields]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
label = 'field' * N
check('literal escaped separator', solve(label + r'\|inside|tail'), [label + '|inside', 'tail'])
check('even escape run leaves separator active', solve(label + '\\\\|tail'), [label + '\\', 'tail'])
check('multiple escaped backslashes', solve('\\\\' * N + '|end'), ['\\' * N, 'end'])
check('empty fields remain positioned', solve('|a||'), ['', 'a', '', ''])
check('empty record is one empty field', solve(''), [''])
check('dangling escape rejected', solve(label + '\\'), None)
check('unknown escape rejected', solve(label + r'\q'), 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:
An escaped pipe splits a field, or an even run of escape characters hides a real separator.

Split a pipe-delimited string. Only backslash-pipe and backslash-backslash are legal escapes. Preserve empty fields; reject incomplete or unknown escapes by returning None.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json
import re
def reject_constant(value):
    raise ValueError('non-JSON numeric constant')
N = 1
observations = []
def solve(text):
    try:
        return {'ok': True, 'value': json.loads(text, parse_constant=reject_constant)}
    except ValueError:
        return {'ok': False}
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('duplicate top-level member', solve('{"a":' + str(N) + ',"a":0}'), {'ok': False})
check('duplicate nested member', solve('{"outer":{"a":1,"a":2}}'), {'ok': False})
check('same name in separate scopes', solve('{"a":' + str(N) + ',"child":{"a":2}}'), {'ok': True, 'value': {'a': N, 'child': {'a': 2}}})
check('escaped name equals literal name', solve(r'{"a":1,"\u0061":2}'), {'ok': False})
check('same member in separate array objects', solve(json.dumps([{'a': i} for i in range(N + 1)])), {'ok': True, 'value': [{'a': i} for i in range(N + 1)]})
check('empty object', solve('{}'), {'ok': True, 'value': {}})
check('valid null differs from decode failure', solve('null'), {'ok': True, 'value': None})
check('malformed syntax', solve('{"a":}'), {'ok': False})
check('non-JSON numeric constant rejected', solve('NaN'), {'ok': 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
import re
def reject_constant(value):
    raise ValueError('non-JSON numeric constant')
N = 1
observations = []
def solve(text):
    keys = re.findall(r'"([^"\\]*)"\s*:', text)
    if len(keys) != len(set(keys)):
        return {'ok': False}
    try:
        return {'ok': True, 'value': json.loads(text, parse_constant=reject_constant)}
    except ValueError:
        return {'ok': False}
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('duplicate top-level member', solve('{"a":' + str(N) + ',"a":0}'), {'ok': False})
check('duplicate nested member', solve('{"outer":{"a":1,"a":2}}'), {'ok': False})
check('same name in separate scopes', solve('{"a":' + str(N) + ',"child":{"a":2}}'), {'ok': True, 'value': {'a': N, 'child': {'a': 2}}})
check('escaped name equals literal name', solve(r'{"a":1,"\u0061":2}'), {'ok': False})
check('same member in separate array objects', solve(json.dumps([{'a': i} for i in range(N + 1)])), {'ok': True, 'value': [{'a': i} for i in range(N + 1)]})
check('empty object', solve('{}'), {'ok': True, 'value': {}})
check('valid null differs from decode failure', solve('null'), {'ok': True, 'value': None})
check('malformed syntax', solve('{"a":}'), {'ok': False})
check('non-JSON numeric constant rejected', solve('NaN'), {'ok': 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:
Conflicting object members collapse to one value, hiding an ambiguous source document.

Decode any valid JSON value into an {ok, value} envelope. Reject syntax errors and duplicate decoded keys at any object depth with {ok: False}; key reuse in different objects is allowed.

split: open-access


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

N = 1
observations = []
def solve(text):
    return (str(len(text)).encode('ascii') + b':' + text.encode('utf-8')).hex()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('ASCII length', solve('x' * N), (str(N).encode() + b':' + b'x' * N).hex())
check('two-byte code point', solve('é' * N), (str(2 * N).encode() + b':' + b'\xc3\xa9' * N).hex())
check('four-byte code point', solve('🙂' * N), (str(4 * N).encode() + b':' + b'\xf0\x9f\x99\x82' * N).hex())
check('combining mark remains separate', solve('e\u0301' * N), (str(3 * N).encode() + b':' + b'e\xcc\x81' * N).hex())
check('mixed encoding widths', solve('A界' * N), (str(4 * N).encode() + b':' + b'A\xe7\x95\x8c' * N).hex())
check('empty payload', solve(''), '303a')
check('delimiter inside payload', solve(':' * N), (str(N).encode() + b':' + b':' * N).hex())
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(text):
    return (str(len(text.encode('utf-16-le')) // 2).encode('ascii') + b':' + text.encode('utf-8')).hex()
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('ASCII length', solve('x' * N), (str(N).encode() + b':' + b'x' * N).hex())
check('two-byte code point', solve('é' * N), (str(2 * N).encode() + b':' + b'\xc3\xa9' * N).hex())
check('four-byte code point', solve('🙂' * N), (str(4 * N).encode() + b':' + b'\xf0\x9f\x99\x82' * N).hex())
check('combining mark remains separate', solve('e\u0301' * N), (str(3 * N).encode() + b':' + b'e\xcc\x81' * N).hex())
check('mixed encoding widths', solve('A界' * N), (str(4 * N).encode() + b':' + b'A\xe7\x95\x8c' * N).hex())
check('empty payload', solve(''), '303a')
check('delimiter inside payload', solve(':' * N), (str(N).encode() + b':' + b':' * N).hex())
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 declared payload length is smaller than the actual UTF-8 payload for non-ASCII text.

Return the hexadecimal representation of an ASCII decimal byte-length prefix, a colon, then the UTF-8 payload. The prefix excludes its own bytes and the colon.

split: open-access