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

N = 1
observations = []
def solve(r):
    return (r['decision_known'] in ['commit','abort']) and (r['transaction_identity'][0] == r['transaction_identity'][1]) and (r['prepare_record'] is True) and (set(r['locks_retained'][0]) <= set(r['locks_retained'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'decision_known': 'commit', 'transaction_identity': ['t2', 't2'], 'prepare_record': True, 'decision_consistent': ['commit', 'commit'], 'locks_retained': [['k1'], ['k1', 'k2']]}
check('valid operation', solve(r), True)
check('An uncertain participant guesses commit without evidence', solve(dict(r, **{'decision_known': 'unknown'})), False)
check('Recovery applies another transaction decision', solve(dict(r, **{'transaction_identity': ['t1', 't2']})), False)
check('Recovery commits a transaction with no durable prepare record', solve(dict(r, **{'prepare_record': False})), False)
check('Recovery reverses a previously recorded final decision', solve(dict(r, **{'decision_consistent': ['abort', 'commit']})), False)
check('Prepared data becomes visible after losing its write locks', solve(dict(r, **{'locks_retained': [['k1'], ['k2']]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'decision_known': 'unknown', 'transaction_identity': ['t1', 't2'], 'prepare_record': False, 'decision_consistent': ['abort', 'commit'], 'locks_retained': [['k1'], ['k2']]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-prepared-recovery

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

N = 1
observations = []
def solve(r):
    return (r['decision_known'] in ['commit','abort']) and (r['transaction_identity'][0] == r['transaction_identity'][1]) and (r['prepare_record'] is True) and (r['decision_consistent'][0] is None or r['decision_consistent'][0] == r['decision_consistent'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'decision_known': 'commit', 'transaction_identity': ['t2', 't2'], 'prepare_record': True, 'decision_consistent': ['commit', 'commit'], 'locks_retained': [['k1'], ['k1', 'k2']]}
check('valid operation', solve(r), True)
check('An uncertain participant guesses commit without evidence', solve(dict(r, **{'decision_known': 'unknown'})), False)
check('Recovery applies another transaction decision', solve(dict(r, **{'transaction_identity': ['t1', 't2']})), False)
check('Recovery commits a transaction with no durable prepare record', solve(dict(r, **{'prepare_record': False})), False)
check('Recovery reverses a previously recorded final decision', solve(dict(r, **{'decision_consistent': ['abort', 'commit']})), False)
check('Prepared data becomes visible after losing its write locks', solve(dict(r, **{'locks_retained': [['k1'], ['k2']]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'decision_known': 'unknown', 'transaction_identity': ['t1', 't2'], 'prepare_record': False, 'decision_consistent': ['abort', 'commit'], 'locks_retained': [['k1'], ['k2']]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The prepared recovery operation is admitted even though recovery reverses a previously recorded final decision.

Return a Boolean admission decision for resolve a prepared participant after recovery. The record r must satisfy all of: r['decision_known'] in ['commit','abort']; r['transaction_identity'][0] == r['transaction_identity'][1]; r['prepare_record'] is True; r['decision_consistent'][0] is None or r['decision_consistent'][0] == r['decision_consistent'][1]; set(r['locks_retained'][0]) <= set(r['locks_retained'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['decision_known'] in ['commit','abort']) and (r['transaction_identity'][0] == r['transaction_identity'][1]) and (r['prepare_record'] is True) and (r['decision_consistent'][0] is None or r['decision_consistent'][0] == r['decision_consistent'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'decision_known': 'commit', 'transaction_identity': ['t2', 't2'], 'prepare_record': True, 'decision_consistent': ['commit', 'commit'], 'locks_retained': [['k1'], ['k1', 'k2']]}
check('valid operation', solve(r), True)
check('An uncertain participant guesses commit without evidence', solve(dict(r, **{'decision_known': 'unknown'})), False)
check('Recovery applies another transaction decision', solve(dict(r, **{'transaction_identity': ['t1', 't2']})), False)
check('Recovery commits a transaction with no durable prepare record', solve(dict(r, **{'prepare_record': False})), False)
check('Recovery reverses a previously recorded final decision', solve(dict(r, **{'decision_consistent': ['abort', 'commit']})), False)
check('Prepared data becomes visible after losing its write locks', solve(dict(r, **{'locks_retained': [['k1'], ['k2']]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'decision_known': 'unknown', 'transaction_identity': ['t1', 't2'], 'prepare_record': False, 'decision_consistent': ['abort', 'commit'], 'locks_retained': [['k1'], ['k2']]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

evaluation_group: xd-prepared-recovery

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

N = 1
observations = []
def solve(r):
    return (r['transaction_identity'][0] == r['transaction_identity'][1]) and (r['prepare_record'] is True) and (r['decision_consistent'][0] is None or r['decision_consistent'][0] == r['decision_consistent'][1]) and (set(r['locks_retained'][0]) <= set(r['locks_retained'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'decision_known': 'commit', 'transaction_identity': ['t2', 't2'], 'prepare_record': True, 'decision_consistent': ['commit', 'commit'], 'locks_retained': [['k1'], ['k1', 'k2']]}
check('valid operation', solve(r), True)
check('An uncertain participant guesses commit without evidence', solve(dict(r, **{'decision_known': 'unknown'})), False)
check('Recovery applies another transaction decision', solve(dict(r, **{'transaction_identity': ['t1', 't2']})), False)
check('Recovery commits a transaction with no durable prepare record', solve(dict(r, **{'prepare_record': False})), False)
check('Recovery reverses a previously recorded final decision', solve(dict(r, **{'decision_consistent': ['abort', 'commit']})), False)
check('Prepared data becomes visible after losing its write locks', solve(dict(r, **{'locks_retained': [['k1'], ['k2']]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'decision_known': 'unknown', 'transaction_identity': ['t1', 't2'], 'prepare_record': False, 'decision_consistent': ['abort', 'commit'], 'locks_retained': [['k1'], ['k2']]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The prepared recovery operation is admitted even though prepared data becomes visible after losing its write locks.

Return a Boolean admission decision for resolve a prepared participant after recovery. The record r must satisfy all of: r['decision_known'] in ['commit','abort']; r['transaction_identity'][0] == r['transaction_identity'][1]; r['prepare_record'] is True; r['decision_consistent'][0] is None or r['decision_consistent'][0] == r['decision_consistent'][1]; set(r['locks_retained'][0]) <= set(r['locks_retained'][1]). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-lease-release

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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The lease release operation is admitted even though one owner releases another owner lease.

Return a Boolean admission decision for release a coordination lease. The record r must satisfy all of: r['owner_match'][0] == r['owner_match'][1]; r['acquisition_match'][0] == r['acquisition_match'][1]; r['resource_match'][0] == r['resource_match'][1]; r['release_not_seen'] is False; r['successor_protected'][0] <= r['successor_protected'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-lease-release

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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The lease release operation is admitted even though a delayed release deletes a later acquisition by the same owner.

Return a Boolean admission decision for release a coordination lease. The record r must satisfy all of: r['owner_match'][0] == r['owner_match'][1]; r['acquisition_match'][0] == r['acquisition_match'][1]; r['resource_match'][0] == r['resource_match'][1]; r['release_not_seen'] is False; r['successor_protected'][0] <= r['successor_protected'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-lease-release

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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The lease release operation is admitted even though a release applies to a different leased resource.

Return a Boolean admission decision for release a coordination lease. The record r must satisfy all of: r['owner_match'][0] == r['owner_match'][1]; r['acquisition_match'][0] == r['acquisition_match'][1]; r['resource_match'][0] == r['resource_match'][1]; r['release_not_seen'] is False; r['successor_protected'][0] <= r['successor_protected'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-lease-release

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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The lease release operation is admitted even though retried release decrements live-lease accounting twice.

Return a Boolean admission decision for release a coordination lease. The record r must satisfy all of: r['owner_match'][0] == r['owner_match'][1]; r['acquisition_match'][0] == r['acquisition_match'][1]; r['resource_match'][0] == r['resource_match'][1]; r['release_not_seen'] is False; r['successor_protected'][0] <= r['successor_protected'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['owner_match'][0] == r['owner_match'][1]) and (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-lease-release

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

N = 1
observations = []
def solve(r):
    return (r['acquisition_match'][0] == r['acquisition_match'][1]) and (r['resource_match'][0] == r['resource_match'][1]) and (r['release_not_seen'] is False) and (r['successor_protected'][0] <= r['successor_protected'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'owner_match': ['a', 'a'], 'acquisition_match': [4, 4], 'resource_match': ['r1', 'r1'], 'release_not_seen': False, 'successor_protected': [4, 4]}
check('valid operation', solve(r), True)
check('One owner releases another owner lease', solve(dict(r, **{'owner_match': ['b', 'a']})), False)
check('A delayed release deletes a later acquisition by the same owner', solve(dict(r, **{'acquisition_match': [3, 4]})), False)
check('A release applies to a different leased resource', solve(dict(r, **{'resource_match': ['r2', 'r1']})), False)
check('Retried release decrements live-lease accounting twice', solve(dict(r, **{'release_not_seen': True})), False)
check('Release removes successor state after ownership transfer', solve(dict(r, **{'successor_protected': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'owner_match': ['b', 'a'], 'acquisition_match': [3, 4], 'resource_match': ['r2', 'r1'], 'release_not_seen': True, 'successor_protected': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The lease release operation is admitted even though release removes successor state after ownership transfer.

Return a Boolean admission decision for release a coordination lease. The record r must satisfy all of: r['owner_match'][0] == r['owner_match'][1]; r['acquisition_match'][0] == r['acquisition_match'][1]; r['resource_match'][0] == r['resource_match'][1]; r['release_not_seen'] is False; r['successor_protected'][0] <= r['successor_protected'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-work-steal

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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The work steal operation is admitted even though a thief executes a task already started by its owner.

Return a Boolean admission decision for steal a task from another worker deque. The record r must satisfy all of: r['task_unstarted'] == 'queued'; r['deque_revision'][0] == r['deque_revision'][1]; set(r['capability_match'][0]) <= set(r['capability_match'][1]); r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]; r['thief_capacity'][0] < r['thief_capacity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-work-steal

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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The work steal operation is admitted even though a thief uses a deque position after the deque changed.

Return a Boolean admission decision for steal a task from another worker deque. The record r must satisfy all of: r['task_unstarted'] == 'queued'; r['deque_revision'][0] == r['deque_revision'][1]; set(r['capability_match'][0]) <= set(r['capability_match'][1]); r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]; r['thief_capacity'][0] < r['thief_capacity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-work-steal

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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The work steal operation is admitted even though a worker steals a task requiring unsupported capabilities.

Return a Boolean admission decision for steal a task from another worker deque. The record r must satisfy all of: r['task_unstarted'] == 'queued'; r['deque_revision'][0] == r['deque_revision'][1]; set(r['capability_match'][0]) <= set(r['capability_match'][1]); r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]; r['thief_capacity'][0] < r['thief_capacity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-work-steal

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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The work steal operation is admitted even though a pinned task leaves its required execution locality.

Return a Boolean admission decision for steal a task from another worker deque. The record r must satisfy all of: r['task_unstarted'] == 'queued'; r['deque_revision'][0] == r['deque_revision'][1]; set(r['capability_match'][0]) <= set(r['capability_match'][1]); r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]; r['thief_capacity'][0] < r['thief_capacity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['task_unstarted'] == 'queued') and (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-work-steal

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

N = 1
observations = []
def solve(r):
    return (r['deque_revision'][0] == r['deque_revision'][1]) and (set(r['capability_match'][0]) <= set(r['capability_match'][1])) and (r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]) and (r['thief_capacity'][0] < r['thief_capacity'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'task_unstarted': 'queued', 'deque_revision': [10, 10], 'capability_match': [['gpu'], ['gpu', 'cpu']], 'locality_constraint': ['z1', 'z1'], 'thief_capacity': [2, 3]}
check('valid operation', solve(r), True)
check('A thief executes a task already started by its owner', solve(dict(r, **{'task_unstarted': 'running'})), False)
check('A thief uses a deque position after the deque changed', solve(dict(r, **{'deque_revision': [9, 10]})), False)
check('A worker steals a task requiring unsupported capabilities', solve(dict(r, **{'capability_match': [['gpu'], ['cpu']]})), False)
check('A pinned task leaves its required execution locality', solve(dict(r, **{'locality_constraint': ['z1', 'z2']})), False)
check('Work stealing exceeds the thief execution slots', solve(dict(r, **{'thief_capacity': [3, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'task_unstarted': 'running', 'deque_revision': [9, 10], 'capability_match': [['gpu'], ['cpu']], 'locality_constraint': ['z1', 'z2'], 'thief_capacity': [3, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The work steal operation is admitted even though work stealing exceeds the thief execution slots.

Return a Boolean admission decision for steal a task from another worker deque. The record r must satisfy all of: r['task_unstarted'] == 'queued'; r['deque_revision'][0] == r['deque_revision'][1]; set(r['capability_match'][0]) <= set(r['capability_match'][1]); r['locality_constraint'][0] is None or r['locality_constraint'][0] == r['locality_constraint'][1]; r['thief_capacity'][0] < r['thief_capacity'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-checkpoint-barrier

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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The checkpoint barrier operation is admitted even though a checkpoint completes before every input channel reaches its barrier.

Return a Boolean admission decision for complete a streaming checkpoint barrier. The record r must satisfy all of: set(r['all_inputs'][0]) == set(r['all_inputs'][1]); len(set(r['barrier_id'])) == 1; r['state_flushed'] is True; r['inflight_captured'][0] == r['inflight_captured'][1]; r['sink_prepared'] == 'prepared'. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-checkpoint-barrier

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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The checkpoint barrier operation is admitted even though barriers from different checkpoint rounds are combined.

Return a Boolean admission decision for complete a streaming checkpoint barrier. The record r must satisfy all of: set(r['all_inputs'][0]) == set(r['all_inputs'][1]); len(set(r['barrier_id'])) == 1; r['state_flushed'] is True; r['inflight_captured'][0] == r['inflight_captured'][1]; r['sink_prepared'] == 'prepared'. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-checkpoint-barrier

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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The checkpoint barrier operation is admitted even though a checkpoint is acknowledged while state exists only in memory.

Return a Boolean admission decision for complete a streaming checkpoint barrier. The record r must satisfy all of: set(r['all_inputs'][0]) == set(r['all_inputs'][1]); len(set(r['barrier_id'])) == 1; r['state_flushed'] is True; r['inflight_captured'][0] == r['inflight_captured'][1]; r['sink_prepared'] == 'prepared'. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-checkpoint-barrier

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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The checkpoint barrier operation is admitted even though unaligned checkpoint omits messages already in transit.

Return a Boolean admission decision for complete a streaming checkpoint barrier. The record r must satisfy all of: set(r['all_inputs'][0]) == set(r['all_inputs'][1]); len(set(r['barrier_id'])) == 1; r['state_flushed'] is True; r['inflight_captured'][0] == r['inflight_captured'][1]; r['sink_prepared'] == 'prepared'. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (set(r['all_inputs'][0]) == set(r['all_inputs'][1])) and (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-checkpoint-barrier

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

N = 1
observations = []
def solve(r):
    return (len(set(r['barrier_id'])) == 1) and (r['state_flushed'] is True) and (r['inflight_captured'][0] == r['inflight_captured'][1]) and (r['sink_prepared'] == 'prepared')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_inputs': [['a', 'b'], ['a', 'b']], 'barrier_id': [4, 4], 'state_flushed': True, 'inflight_captured': [8, 8], 'sink_prepared': 'prepared'}
check('valid operation', solve(r), True)
check('A checkpoint completes before every input channel reaches its barrier', solve(dict(r, **{'all_inputs': [['a', 'b'], ['a']]})), False)
check('Barriers from different checkpoint rounds are combined', solve(dict(r, **{'barrier_id': [4, 5]})), False)
check('A checkpoint is acknowledged while state exists only in memory', solve(dict(r, **{'state_flushed': False})), False)
check('Unaligned checkpoint omits messages already in transit', solve(dict(r, **{'inflight_captured': [5, 8]})), False)
check('Upstream progress commits before the transactional sink prepares', solve(dict(r, **{'sink_prepared': 'open'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_inputs': [['a', 'b'], ['a']], 'barrier_id': [4, 5], 'state_flushed': False, 'inflight_captured': [5, 8], 'sink_prepared': 'open'}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The checkpoint barrier operation is admitted even though upstream progress commits before the transactional sink prepares.

Return a Boolean admission decision for complete a streaming checkpoint barrier. The record r must satisfy all of: set(r['all_inputs'][0]) == set(r['all_inputs'][1]); len(set(r['barrier_id'])) == 1; r['state_flushed'] is True; r['inflight_captured'][0] == r['inflight_captured'][1]; r['sink_prepared'] == 'prepared'. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['source_writes_stopped'] is True) and (all(r['dependency_ready'])) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-regional-cutover

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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (all(r['dependency_ready'])) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The regional cutover operation is admitted even though cutover loses writes beyond the destination applied watermark.

Return a Boolean admission decision for cut traffic to a replacement region. The record r must satisfy all of: r['replication_caught_up'][0] >= r['replication_caught_up'][1]; r['source_writes_stopped'] is True; all(r['dependency_ready']); r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]; r['cutover_generation'][0] > r['cutover_generation'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (all(r['dependency_ready'])) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-regional-cutover

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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (r['source_writes_stopped'] is True) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The regional cutover operation is admitted even though both regions accept writes during a single-writer cutover.

Return a Boolean admission decision for cut traffic to a replacement region. The record r must satisfy all of: r['replication_caught_up'][0] >= r['replication_caught_up'][1]; r['source_writes_stopped'] is True; all(r['dependency_ready']); r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]; r['cutover_generation'][0] > r['cutover_generation'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (r['source_writes_stopped'] is True) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-regional-cutover

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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (r['source_writes_stopped'] is True) and (all(r['dependency_ready'])) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The regional cutover operation is admitted even though application traffic arrives before regional dependencies are ready.

Return a Boolean admission decision for cut traffic to a replacement region. The record r must satisfy all of: r['replication_caught_up'][0] >= r['replication_caught_up'][1]; r['source_writes_stopped'] is True; all(r['dependency_ready']); r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]; r['cutover_generation'][0] > r['cutover_generation'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (r['source_writes_stopped'] is True) and (all(r['dependency_ready'])) and (r['cutover_generation'][0] > r['cutover_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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)

evaluation_group: xd-regional-cutover

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

N = 1
observations = []
def solve(r):
    return (r['replication_caught_up'][0] >= r['replication_caught_up'][1]) and (r['source_writes_stopped'] is True) and (all(r['dependency_ready'])) and (r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'replication_caught_up': [30, 30], 'source_writes_stopped': True, 'dependency_ready': [True, True], 'capacity_headroom': [4, 5, 10], 'cutover_generation': [8, 7]}
check('valid operation', solve(r), True)
check('Cutover loses writes beyond the destination applied watermark', solve(dict(r, **{'replication_caught_up': [29, 30]})), False)
check('Both regions accept writes during a single-writer cutover', solve(dict(r, **{'source_writes_stopped': False})), False)
check('Application traffic arrives before regional dependencies are ready', solve(dict(r, **{'dependency_ready': [True, False]})), False)
check('Destination capacity cannot serve the transferred load', solve(dict(r, **{'capacity_headroom': [6, 5, 10]})), False)
check('A delayed cutover command reverses a newer regional decision', solve(dict(r, **{'cutover_generation': [7, 7]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'replication_caught_up': [29, 30], 'source_writes_stopped': False, 'dependency_ready': [True, False], 'capacity_headroom': [6, 5, 10], 'cutover_generation': [7, 7]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), 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:
The regional cutover operation is admitted even though destination capacity cannot serve the transferred load.

Return a Boolean admission decision for cut traffic to a replacement region. The record r must satisfy all of: r['replication_caught_up'][0] >= r['replication_caught_up'][1]; r['source_writes_stopped'] is True; all(r['dependency_ready']); r['capacity_headroom'][0]+r['capacity_headroom'][1] <= r['capacity_headroom'][2]; r['cutover_generation'][0] > r['cutover_generation'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access