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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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-workflow-cancel

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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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 workflow cancel operation is admitted even though cancellation rewrites an already successful terminal result.

Return a Boolean admission decision for complete workflow cancellation. The record r must satisfy all of: all(r['child_propagation']); r['activity_cleanup'] is True; r['terminal_stability'] not in ['completed','failed']; r['request_generation'][0] == r['request_generation'][1]; r['cleanup_deadline'][0] <= r['cleanup_deadline'][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 (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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-workflow-cancel

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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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 workflow cancel operation is admitted even though an old cancellation request terminates a newer run.

Return a Boolean admission decision for complete workflow cancellation. The record r must satisfy all of: all(r['child_propagation']); r['activity_cleanup'] is True; r['terminal_stability'] not in ['completed','failed']; r['request_generation'][0] == r['request_generation'][1]; r['cleanup_deadline'][0] <= r['cleanup_deadline'][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 (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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-workflow-cancel

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

N = 1
observations = []
def solve(r):
    return (r['activity_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
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 workflow cancel operation is admitted even though cancellation waits beyond its explicit cleanup deadline.

Return a Boolean admission decision for complete workflow cancellation. The record r must satisfy all of: all(r['child_propagation']); r['activity_cleanup'] is True; r['terminal_stability'] not in ['completed','failed']; r['request_generation'][0] == r['request_generation'][1]; r['cleanup_deadline'][0] <= r['cleanup_deadline'][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 (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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-workflow-join

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

N = 1
observations = []
def solve(r):
    return (set(r['all_required'][0]) <= set(r['all_required'][1])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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 workflow join operation is admitted even though a join proceeds before every required predecessor completes.

Return a Boolean admission decision for release a workflow dependency join. The record r must satisfy all of: set(r['all_required'][0]) <= set(r['all_required'][1]); all(x == 'success' for x in r['success_policy']); all(x == r['branch_generation'][0] for x in r['branch_generation'][1]); r['release_once'] is False; not any(r['cancelled_branch']). 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_required'][0]) <= set(r['all_required'][1])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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-workflow-join

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

N = 1
observations = []
def solve(r):
    return (set(r['all_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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 workflow join operation is admitted even though a success-only join treats failed predecessors as successful.

Return a Boolean admission decision for release a workflow dependency join. The record r must satisfy all of: set(r['all_required'][0]) <= set(r['all_required'][1]); all(x == 'success' for x in r['success_policy']); all(x == r['branch_generation'][0] for x in r['branch_generation'][1]); r['release_once'] is False; not any(r['cancelled_branch']). 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_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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-workflow-join

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

N = 1
observations = []
def solve(r):
    return (set(r['all_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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 workflow join operation is admitted even though a completion from a previous loop iteration satisfies the join.

Return a Boolean admission decision for release a workflow dependency join. The record r must satisfy all of: set(r['all_required'][0]) <= set(r['all_required'][1]); all(x == 'success' for x in r['success_policy']); all(x == r['branch_generation'][0] for x in r['branch_generation'][1]); r['release_once'] is False; not any(r['cancelled_branch']). 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_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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-workflow-join

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

N = 1
observations = []
def solve(r):
    return (set(r['all_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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 workflow join operation is admitted even though duplicate predecessor events release the join twice.

Return a Boolean admission decision for release a workflow dependency join. The record r must satisfy all of: set(r['all_required'][0]) <= set(r['all_required'][1]); all(x == 'success' for x in r['success_policy']); all(x == r['branch_generation'][0] for x in r['branch_generation'][1]); r['release_once'] is False; not any(r['cancelled_branch']). 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_required'][0]) <= set(r['all_required'][1])) and (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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-workflow-join

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

N = 1
observations = []
def solve(r):
    return (all(x == 'success' for x in r['success_policy'])) and (all(x == r['branch_generation'][0] for x in r['branch_generation'][1])) and (r['release_once'] is False) and (not any(r['cancelled_branch']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'all_required': [['a', 'b'], ['a', 'b']], 'success_policy': ['success', 'success'], 'branch_generation': [3, [3, 3]], 'release_once': False, 'cancelled_branch': [False, False]}
check('valid operation', solve(r), True)
check('A join proceeds before every required predecessor completes', solve(dict(r, **{'all_required': [['a', 'b'], ['a']]})), False)
check('A success-only join treats failed predecessors as successful', solve(dict(r, **{'success_policy': ['success', 'failed']})), False)
check('A completion from a previous loop iteration satisfies the join', solve(dict(r, **{'branch_generation': [3, [2, 3]]})), False)
check('Duplicate predecessor events release the join twice', solve(dict(r, **{'release_once': True})), False)
check('A cancelled required branch is silently omitted', solve(dict(r, **{'cancelled_branch': [False, True]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'all_required': [['a', 'b'], ['a']], 'success_policy': ['success', 'failed'], 'branch_generation': [3, [2, 3]], 'release_once': True, 'cancelled_branch': [False, True]}
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 workflow join operation is admitted even though a cancelled required branch is silently omitted.

Return a Boolean admission decision for release a workflow dependency join. The record r must satisfy all of: set(r['all_required'][0]) <= set(r['all_required'][1]); all(x == 'success' for x in r['success_policy']); all(x == r['branch_generation'][0] for x in r['branch_generation'][1]); r['release_once'] is False; not any(r['cancelled_branch']). 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['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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-workflow-heartbeat

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

N = 1
observations = []
def solve(r):
    return (r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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 workflow heartbeat operation is admitted even though a previous activity attempt keeps the new attempt alive.

Return a Boolean admission decision for renew a durable activity heartbeat. The record r must satisfy all of: r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]; r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]; r['progress_monotonic'][0] >= r['progress_monotonic'][1]; len(r['payload_limit'][0]) <= r['payload_limit'][1]; r['absolute_deadline'][0] <= r['absolute_deadline'][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['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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-workflow-heartbeat

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

N = 1
observations = []
def solve(r):
    return (r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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 workflow heartbeat operation is admitted even though a heartbeat resurrects an activity after its timeout.

Return a Boolean admission decision for renew a durable activity heartbeat. The record r must satisfy all of: r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]; r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]; r['progress_monotonic'][0] >= r['progress_monotonic'][1]; len(r['payload_limit'][0]) <= r['payload_limit'][1]; r['absolute_deadline'][0] <= r['absolute_deadline'][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['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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-workflow-heartbeat

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

N = 1
observations = []
def solve(r):
    return (r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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 workflow heartbeat operation is admitted even though a late heartbeat regresses the resumable progress marker.

Return a Boolean admission decision for renew a durable activity heartbeat. The record r must satisfy all of: r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]; r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]; r['progress_monotonic'][0] >= r['progress_monotonic'][1]; len(r['payload_limit'][0]) <= r['payload_limit'][1]; r['absolute_deadline'][0] <= r['absolute_deadline'][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['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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-workflow-heartbeat

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

N = 1
observations = []
def solve(r):
    return (r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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 workflow heartbeat operation is admitted even though heartbeat detail exceeds durable history payload limits.

Return a Boolean admission decision for renew a durable activity heartbeat. The record r must satisfy all of: r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]; r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]; r['progress_monotonic'][0] >= r['progress_monotonic'][1]; len(r['payload_limit'][0]) <= r['payload_limit'][1]; r['absolute_deadline'][0] <= r['absolute_deadline'][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['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]) and (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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-workflow-heartbeat

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

N = 1
observations = []
def solve(r):
    return (r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]) and (r['progress_monotonic'][0] >= r['progress_monotonic'][1]) and (len(r['payload_limit'][0]) <= r['payload_limit'][1]) and (r['absolute_deadline'][0] <= r['absolute_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'heartbeat_attempt': [4, 4], 'heartbeat_expiry': [9, 10], 'progress_monotonic': [20, 19], 'payload_limit': ['abc', 3], 'absolute_deadline': [20, 20]}
check('valid operation', solve(r), True)
check('A previous activity attempt keeps the new attempt alive', solve(dict(r, **{'heartbeat_attempt': [3, 4]})), False)
check('A heartbeat resurrects an activity after its timeout', solve(dict(r, **{'heartbeat_expiry': [10, 10]})), False)
check('A late heartbeat regresses the resumable progress marker', solve(dict(r, **{'progress_monotonic': [18, 19]})), False)
check('Heartbeat detail exceeds durable history payload limits', solve(dict(r, **{'payload_limit': ['abcd', 3]})), False)
check('Heartbeat renewal extends beyond schedule-to-close timeout', solve(dict(r, **{'absolute_deadline': [21, 20]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'heartbeat_attempt': [3, 4], 'heartbeat_expiry': [10, 10], 'progress_monotonic': [18, 19], 'payload_limit': ['abcd', 3], 'absolute_deadline': [21, 20]}
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 workflow heartbeat operation is admitted even though heartbeat renewal extends beyond schedule-to-close timeout.

Return a Boolean admission decision for renew a durable activity heartbeat. The record r must satisfy all of: r['heartbeat_attempt'][0] == r['heartbeat_attempt'][1]; r['heartbeat_expiry'][0] < r['heartbeat_expiry'][1]; r['progress_monotonic'][0] >= r['progress_monotonic'][1]; len(r['payload_limit'][0]) <= r['payload_limit'][1]; r['absolute_deadline'][0] <= r['absolute_deadline'][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['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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-distributed-lock

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

N = 1
observations = []
def solve(r):
    return (r['lock_available'] is None) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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 distributed lock operation is admitted even though acquisition steals a currently held mutex.

Return a Boolean admission decision for acquire a distributed mutex. The record r must satisfy all of: r['lock_available'] is None; r['waiter_turn'][0] == r['waiter_turn'][1][0]; r['owner_session'][0] < r['owner_session'][1]; r['fence_increase'][0] > r['fence_increase'][1]; r['lock_namespace'][0] == r['lock_namespace'][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['lock_available'] is None) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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-distributed-lock

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

N = 1
observations = []
def solve(r):
    return (r['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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 distributed lock operation is admitted even though a newcomer bypasses the oldest eligible waiter.

Return a Boolean admission decision for acquire a distributed mutex. The record r must satisfy all of: r['lock_available'] is None; r['waiter_turn'][0] == r['waiter_turn'][1][0]; r['owner_session'][0] < r['owner_session'][1]; r['fence_increase'][0] > r['fence_increase'][1]; r['lock_namespace'][0] == r['lock_namespace'][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['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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-distributed-lock

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

N = 1
observations = []
def solve(r):
    return (r['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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 distributed lock operation is admitted even though an expired session acquires a new mutex.

Return a Boolean admission decision for acquire a distributed mutex. The record r must satisfy all of: r['lock_available'] is None; r['waiter_turn'][0] == r['waiter_turn'][1][0]; r['owner_session'][0] < r['owner_session'][1]; r['fence_increase'][0] > r['fence_increase'][1]; r['lock_namespace'][0] == r['lock_namespace'][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['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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-distributed-lock

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

N = 1
observations = []
def solve(r):
    return (r['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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 distributed lock operation is admitted even though reacquisition reuses the previous fencing token.

Return a Boolean admission decision for acquire a distributed mutex. The record r must satisfy all of: r['lock_available'] is None; r['waiter_turn'][0] == r['waiter_turn'][1][0]; r['owner_session'][0] < r['owner_session'][1]; r['fence_increase'][0] > r['fence_increase'][1]; r['lock_namespace'][0] == r['lock_namespace'][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['lock_available'] is None) and (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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-distributed-lock

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

N = 1
observations = []
def solve(r):
    return (r['waiter_turn'][0] == r['waiter_turn'][1][0]) and (r['owner_session'][0] < r['owner_session'][1]) and (r['fence_increase'][0] > r['fence_increase'][1]) and (r['lock_namespace'][0] == r['lock_namespace'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'lock_available': None, 'waiter_turn': ['a', ['a', 'b']], 'owner_session': [9, 10], 'fence_increase': [7, 6], 'lock_namespace': ['n1', 'n1']}
check('valid operation', solve(r), True)
check('Acquisition steals a currently held mutex', solve(dict(r, **{'lock_available': 'owner'})), False)
check('A newcomer bypasses the oldest eligible waiter', solve(dict(r, **{'waiter_turn': ['b', ['a', 'b']]})), False)
check('An expired session acquires a new mutex', solve(dict(r, **{'owner_session': [10, 10]})), False)
check('Reacquisition reuses the previous fencing token', solve(dict(r, **{'fence_increase': [6, 6]})), False)
check('A request acquires a lock in another coordination namespace', solve(dict(r, **{'lock_namespace': ['n2', 'n1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'lock_available': 'owner', 'waiter_turn': ['b', ['a', 'b']], 'owner_session': [10, 10], 'fence_increase': [6, 6], 'lock_namespace': ['n2', 'n1']}
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 distributed lock operation is admitted even though a request acquires a lock in another coordination namespace.

Return a Boolean admission decision for acquire a distributed mutex. The record r must satisfy all of: r['lock_available'] is None; r['waiter_turn'][0] == r['waiter_turn'][1][0]; r['owner_session'][0] < r['owner_session'][1]; r['fence_increase'][0] > r['fence_increase'][1]; r['lock_namespace'][0] == r['lock_namespace'][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['positive_request'] > 0) and (r['request_not_granted'][0] not in r['request_not_granted'][1]) and (r['weighted_fifo'][0] == r['weighted_fifo'][1][0]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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-semaphore-grant

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

N = 1
observations = []
def solve(r):
    return (r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]) and (r['request_not_granted'][0] not in r['request_not_granted'][1]) and (r['weighted_fifo'][0] == r['weighted_fifo'][1][0]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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 semaphore grant operation is admitted even though the grant exceeds the configured total permit count.

Return a Boolean admission decision for grant permits from a distributed semaphore. The record r must satisfy all of: r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]; r['positive_request'] > 0; r['request_not_granted'][0] not in r['request_not_granted'][1]; r['weighted_fifo'][0] == r['weighted_fifo'][1][0]; r['grant_session'] in ['active']. 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['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]) and (r['request_not_granted'][0] not in r['request_not_granted'][1]) and (r['weighted_fifo'][0] == r['weighted_fifo'][1][0]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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-semaphore-grant

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

N = 1
observations = []
def solve(r):
    return (r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]) and (r['positive_request'] > 0) and (r['weighted_fifo'][0] == r['weighted_fifo'][1][0]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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 semaphore grant operation is admitted even though a zero or negative grant corrupts permit accounting.

Return a Boolean admission decision for grant permits from a distributed semaphore. The record r must satisfy all of: r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]; r['positive_request'] > 0; r['request_not_granted'][0] not in r['request_not_granted'][1]; r['weighted_fifo'][0] == r['weighted_fifo'][1][0]; r['grant_session'] in ['active']. 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['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]) and (r['positive_request'] > 0) and (r['weighted_fifo'][0] == r['weighted_fifo'][1][0]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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-semaphore-grant

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

N = 1
observations = []
def solve(r):
    return (r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]) and (r['positive_request'] > 0) and (r['request_not_granted'][0] not in r['request_not_granted'][1]) and (r['grant_session'] in ['active'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'permit_capacity': [3, 2, 5], 'positive_request': 1, 'request_not_granted': ['r2', ['r1']], 'weighted_fifo': ['big', ['big', 'small']], 'grant_session': 'active'}
check('valid operation', solve(r), True)
check('The grant exceeds the configured total permit count', solve(dict(r, **{'permit_capacity': [4, 2, 5]})), False)
check('A zero or negative grant corrupts permit accounting', solve(dict(r, **{'positive_request': 0})), False)
check('A retried grant allocates the same request twice', solve(dict(r, **{'request_not_granted': ['r1', ['r1']]})), False)
check('A later small request overtakes the queued weighted head', solve(dict(r, **{'weighted_fifo': ['small', ['big', 'small']]})), False)
check('A session destroyed while waiting receives permits', solve(dict(r, **{'grant_session': 'closed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'permit_capacity': [4, 2, 5], 'positive_request': 0, 'request_not_granted': ['r1', ['r1']], 'weighted_fifo': ['small', ['big', 'small']], 'grant_session': 'closed'}
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 semaphore grant operation is admitted even though a retried grant allocates the same request twice.

Return a Boolean admission decision for grant permits from a distributed semaphore. The record r must satisfy all of: r['permit_capacity'][0]+r['permit_capacity'][1] <= r['permit_capacity'][2]; r['positive_request'] > 0; r['request_not_granted'][0] not in r['request_not_granted'][1]; r['weighted_fifo'][0] == r['weighted_fifo'][1][0]; r['grant_session'] in ['active']. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access