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

N = 1
observations = []
def solve(r):
    return (r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]) and (r['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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-circuit-probe

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

N = 1
observations = []
def solve(r):
    return (r['probe_exclusive'] == 0) and (r['circuit_generation'][0] == r['circuit_generation'][1]) and (r['endpoint_health'][0] in r['endpoint_health'][1]) and (r['caller_deadline'][0] < r['caller_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'open_delay': [10, 5, 5], 'probe_exclusive': 0, 'circuit_generation': [4, 4], 'endpoint_health': ['a', ['a', 'b']], 'caller_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A probe runs before the open-state cooldown elapses', solve(dict(r, **{'open_delay': [9, 5, 5]})), False)
check('Concurrent requests occupy the single half-open probe slot', solve(dict(r, **{'probe_exclusive': 1})), False)
check('A stale timer probes a newly opened circuit generation', solve(dict(r, **{'circuit_generation': [3, 4]})), False)
check('A probe targets an endpoint removed from service discovery', solve(dict(r, **{'endpoint_health': ['c', ['a', 'b']]})), False)
check('A probe is started for an already expired caller', solve(dict(r, **{'caller_deadline': [11, 11]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'open_delay': [9, 5, 5], 'probe_exclusive': 1, 'circuit_generation': [3, 4], 'endpoint_health': ['c', ['a', 'b']], 'caller_deadline': [11, 11]}
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 circuit probe operation is admitted even though a probe is started for an already expired caller.

Return a Boolean admission decision for allow a half-open circuit probe. The record r must satisfy all of: r['open_delay'][0]-r['open_delay'][1] >= r['open_delay'][2]; r['probe_exclusive'] == 0; r['circuit_generation'][0] == r['circuit_generation'][1]; r['endpoint_health'][0] in r['endpoint_health'][1]; r['caller_deadline'][0] < r['caller_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['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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-hedge-request

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

N = 1
observations = []
def solve(r):
    return (r['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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 hedge request operation is admitted even though a duplicate request is launched before the hedge delay.

Return a Boolean admission decision for launch a speculative hedged request. The record r must satisfy all of: r['hedge_delay'][0] >= r['hedge_delay'][1]; r['primary_pending'] == 'pending'; r['separate_domain'][0] != r['separate_domain'][1]; r['hedge_budget'][0] < r['hedge_budget'][1]; r['operation_repeatable'] is True. 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['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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-hedge-request

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

N = 1
observations = []
def solve(r):
    return (r['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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 hedge request operation is admitted even though a hedge starts after the primary already completed.

Return a Boolean admission decision for launch a speculative hedged request. The record r must satisfy all of: r['hedge_delay'][0] >= r['hedge_delay'][1]; r['primary_pending'] == 'pending'; r['separate_domain'][0] != r['separate_domain'][1]; r['hedge_budget'][0] < r['hedge_budget'][1]; r['operation_repeatable'] is True. 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['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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-hedge-request

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

N = 1
observations = []
def solve(r):
    return (r['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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 hedge request operation is admitted even though primary and hedge share the same correlated failure domain.

Return a Boolean admission decision for launch a speculative hedged request. The record r must satisfy all of: r['hedge_delay'][0] >= r['hedge_delay'][1]; r['primary_pending'] == 'pending'; r['separate_domain'][0] != r['separate_domain'][1]; r['hedge_budget'][0] < r['hedge_budget'][1]; r['operation_repeatable'] is True. 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['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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-hedge-request

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

N = 1
observations = []
def solve(r):
    return (r['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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 hedge request operation is admitted even though speculation consumes more than the allowed duplicate capacity.

Return a Boolean admission decision for launch a speculative hedged request. The record r must satisfy all of: r['hedge_delay'][0] >= r['hedge_delay'][1]; r['primary_pending'] == 'pending'; r['separate_domain'][0] != r['separate_domain'][1]; r['hedge_budget'][0] < r['hedge_budget'][1]; r['operation_repeatable'] is True. 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['hedge_delay'][0] >= r['hedge_delay'][1]) and (r['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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-hedge-request

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

N = 1
observations = []
def solve(r):
    return (r['primary_pending'] == 'pending') and (r['separate_domain'][0] != r['separate_domain'][1]) and (r['hedge_budget'][0] < r['hedge_budget'][1]) and (r['operation_repeatable'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'hedge_delay': [5, 5], 'primary_pending': 'pending', 'separate_domain': ['z1', 'z2'], 'hedge_budget': [0, 1], 'operation_repeatable': True}
check('valid operation', solve(r), True)
check('A duplicate request is launched before the hedge delay', solve(dict(r, **{'hedge_delay': [4, 5]})), False)
check('A hedge starts after the primary already completed', solve(dict(r, **{'primary_pending': 'done'})), False)
check('Primary and hedge share the same correlated failure domain', solve(dict(r, **{'separate_domain': ['z1', 'z1']})), False)
check('Speculation consumes more than the allowed duplicate capacity', solve(dict(r, **{'hedge_budget': [1, 1]})), False)
check('A hedge duplicates a non-repeatable side effect', solve(dict(r, **{'operation_repeatable': False})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'hedge_delay': [4, 5], 'primary_pending': 'done', 'separate_domain': ['z1', 'z1'], 'hedge_budget': [1, 1], 'operation_repeatable': False}
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 hedge request operation is admitted even though a hedge duplicates a non-repeatable side effect.

Return a Boolean admission decision for launch a speculative hedged request. The record r must satisfy all of: r['hedge_delay'][0] >= r['hedge_delay'][1]; r['primary_pending'] == 'pending'; r['separate_domain'][0] != r['separate_domain'][1]; r['hedge_budget'][0] < r['hedge_budget'][1]; r['operation_repeatable'] is True. 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['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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-bulkhead-admission

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

N = 1
observations = []
def solve(r):
    return (r['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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 bulkhead admission operation is admitted even though concurrent calls exceed the dependency execution limit.

Return a Boolean admission decision for enter a dependency bulkhead. The record r must satisfy all of: r['active_capacity'][0] < r['active_capacity'][1]; r['queue_capacity'][0] < r['queue_capacity'][1]; r['queue_deadline'][0] < r['queue_deadline'][1]; r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]; r['isolation_key'][0] == r['isolation_key'][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['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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-bulkhead-admission

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

N = 1
observations = []
def solve(r):
    return (r['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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 bulkhead admission operation is admitted even though waiting callers exceed the bounded waiting queue.

Return a Boolean admission decision for enter a dependency bulkhead. The record r must satisfy all of: r['active_capacity'][0] < r['active_capacity'][1]; r['queue_capacity'][0] < r['queue_capacity'][1]; r['queue_deadline'][0] < r['queue_deadline'][1]; r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]; r['isolation_key'][0] == r['isolation_key'][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['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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-bulkhead-admission

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

N = 1
observations = []
def solve(r):
    return (r['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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 bulkhead admission operation is admitted even though a waiting call starts after its queue deadline.

Return a Boolean admission decision for enter a dependency bulkhead. The record r must satisfy all of: r['active_capacity'][0] < r['active_capacity'][1]; r['queue_capacity'][0] < r['queue_capacity'][1]; r['queue_deadline'][0] < r['queue_deadline'][1]; r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]; r['isolation_key'][0] == r['isolation_key'][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['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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-bulkhead-admission

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

N = 1
observations = []
def solve(r):
    return (r['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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 bulkhead admission operation is admitted even though normal traffic consumes capacity reserved for recovery.

Return a Boolean admission decision for enter a dependency bulkhead. The record r must satisfy all of: r['active_capacity'][0] < r['active_capacity'][1]; r['queue_capacity'][0] < r['queue_capacity'][1]; r['queue_deadline'][0] < r['queue_deadline'][1]; r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]; r['isolation_key'][0] == r['isolation_key'][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['active_capacity'][0] < r['active_capacity'][1]) and (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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-bulkhead-admission

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

N = 1
observations = []
def solve(r):
    return (r['queue_capacity'][0] < r['queue_capacity'][1]) and (r['queue_deadline'][0] < r['queue_deadline'][1]) and (r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]) and (r['isolation_key'][0] == r['isolation_key'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'active_capacity': [2, 3], 'queue_capacity': [4, 5], 'queue_deadline': [8, 9], 'priority_reservation': [False, 2, 5, 2], 'isolation_key': ['db1', 'db1']}
check('valid operation', solve(r), True)
check('Concurrent calls exceed the dependency execution limit', solve(dict(r, **{'active_capacity': [3, 3]})), False)
check('Waiting callers exceed the bounded waiting queue', solve(dict(r, **{'queue_capacity': [5, 5]})), False)
check('A waiting call starts after its queue deadline', solve(dict(r, **{'queue_deadline': [9, 9]})), False)
check('Normal traffic consumes capacity reserved for recovery', solve(dict(r, **{'priority_reservation': [False, 3, 5, 2]})), False)
check('Unrelated dependencies share a bulkhead accounting key', solve(dict(r, **{'isolation_key': ['db2', 'db1']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'active_capacity': [3, 3], 'queue_capacity': [5, 5], 'queue_deadline': [9, 9], 'priority_reservation': [False, 3, 5, 2], 'isolation_key': ['db2', 'db1']}
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 bulkhead admission operation is admitted even though unrelated dependencies share a bulkhead accounting key.

Return a Boolean admission decision for enter a dependency bulkhead. The record r must satisfy all of: r['active_capacity'][0] < r['active_capacity'][1]; r['queue_capacity'][0] < r['queue_capacity'][1]; r['queue_deadline'][0] < r['queue_deadline'][1]; r['priority_reservation'][0] or r['priority_reservation'][1] < r['priority_reservation'][2]-r['priority_reservation'][3]; r['isolation_key'][0] == r['isolation_key'][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['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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-health-eviction

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

N = 1
observations = []
def solve(r):
    return (r['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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 health eviction operation is admitted even though one intermittent failure causes premature eviction.

Return a Boolean admission decision for evict an unhealthy backend. The record r must satisfy all of: r['consecutive_failures'][0] >= r['consecutive_failures'][1]; r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]; r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]; (r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]; r['backend_incarnation'][0] == r['backend_incarnation'][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['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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-health-eviction

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

N = 1
observations = []
def solve(r):
    return (r['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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 health eviction operation is admitted even though eviction removes the last required healthy capacity.

Return a Boolean admission decision for evict an unhealthy backend. The record r must satisfy all of: r['consecutive_failures'][0] >= r['consecutive_failures'][1]; r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]; r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]; (r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]; r['backend_incarnation'][0] == r['backend_incarnation'][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['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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-health-eviction

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

N = 1
observations = []
def solve(r):
    return (r['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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 health eviction operation is admitted even though old failure samples evict a recovered backend.

Return a Boolean admission decision for evict an unhealthy backend. The record r must satisfy all of: r['consecutive_failures'][0] >= r['consecutive_failures'][1]; r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]; r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]; (r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]; r['backend_incarnation'][0] == r['backend_incarnation'][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['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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-health-eviction

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

N = 1
observations = []
def solve(r):
    return (r['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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 health eviction operation is admitted even though correlated failures evict more than the allowed pool fraction.

Return a Boolean admission decision for evict an unhealthy backend. The record r must satisfy all of: r['consecutive_failures'][0] >= r['consecutive_failures'][1]; r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]; r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]; (r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]; r['backend_incarnation'][0] == r['backend_incarnation'][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['consecutive_failures'][0] >= r['consecutive_failures'][1]) and (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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-health-eviction

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

N = 1
observations = []
def solve(r):
    return (r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]) and (r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]) and ((r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]) and (r['backend_incarnation'][0] == r['backend_incarnation'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'consecutive_failures': [3, 3], 'minimum_survivors': [3, 2], 'sample_freshness': [10, 8, 2], 'ejection_limit': [1, 10, 20], 'backend_incarnation': ['p2', 'p2']}
check('valid operation', solve(r), True)
check('One intermittent failure causes premature eviction', solve(dict(r, **{'consecutive_failures': [2, 3]})), False)
check('Eviction removes the last required healthy capacity', solve(dict(r, **{'minimum_survivors': [2, 2]})), False)
check('Old failure samples evict a recovered backend', solve(dict(r, **{'sample_freshness': [10, 7, 2]})), False)
check('Correlated failures evict more than the allowed pool fraction', solve(dict(r, **{'ejection_limit': [2, 10, 20]})), False)
check('A failed probe evicts a replacement process at the same address', solve(dict(r, **{'backend_incarnation': ['p1', 'p2']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'consecutive_failures': [2, 3], 'minimum_survivors': [2, 2], 'sample_freshness': [10, 7, 2], 'ejection_limit': [2, 10, 20], 'backend_incarnation': ['p1', 'p2']}
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 health eviction operation is admitted even though a failed probe evicts a replacement process at the same address.

Return a Boolean admission decision for evict an unhealthy backend. The record r must satisfy all of: r['consecutive_failures'][0] >= r['consecutive_failures'][1]; r['minimum_survivors'][0]-1 >= r['minimum_survivors'][1]; r['sample_freshness'][0]-r['sample_freshness'][1] <= r['sample_freshness'][2]; (r['ejection_limit'][0]+1)*100 <= r['ejection_limit'][1]*r['ejection_limit'][2]; r['backend_incarnation'][0] == r['backend_incarnation'][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['inflight_zero'] == 0) and (r['registration_removed'] is False) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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-graceful-drain

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

N = 1
observations = []
def solve(r):
    return (r['new_admission_closed'] is False) and (r['registration_removed'] is False) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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 graceful drain operation is admitted even though shutdown still permits new work to enter.

Return a Boolean admission decision for finish draining a service instance. The record r must satisfy all of: r['new_admission_closed'] is False; r['inflight_zero'] == 0; r['registration_removed'] is False; r['flush_complete'] is True; all(r['children_reaped']). 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['new_admission_closed'] is False) and (r['registration_removed'] is False) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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-graceful-drain

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

N = 1
observations = []
def solve(r):
    return (r['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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 graceful drain operation is admitted even though shutdown terminates in-flight requests prematurely.

Return a Boolean admission decision for finish draining a service instance. The record r must satisfy all of: r['new_admission_closed'] is False; r['inflight_zero'] == 0; r['registration_removed'] is False; r['flush_complete'] is True; all(r['children_reaped']). 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['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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-graceful-drain

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

N = 1
observations = []
def solve(r):
    return (r['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['registration_removed'] is False) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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 graceful drain operation is admitted even though shutdown completes while routing still advertises the instance.

Return a Boolean admission decision for finish draining a service instance. The record r must satisfy all of: r['new_admission_closed'] is False; r['inflight_zero'] == 0; r['registration_removed'] is False; r['flush_complete'] is True; all(r['children_reaped']). 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['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['registration_removed'] is False) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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-graceful-drain

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

N = 1
observations = []
def solve(r):
    return (r['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['registration_removed'] is False) and (r['flush_complete'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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 graceful drain operation is admitted even though shutdown discards buffered durable work.

Return a Boolean admission decision for finish draining a service instance. The record r must satisfy all of: r['new_admission_closed'] is False; r['inflight_zero'] == 0; r['registration_removed'] is False; r['flush_complete'] is True; all(r['children_reaped']). 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['new_admission_closed'] is False) and (r['inflight_zero'] == 0) and (r['registration_removed'] is False) and (r['flush_complete'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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-graceful-drain

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

N = 1
observations = []
def solve(r):
    return (r['inflight_zero'] == 0) and (r['registration_removed'] is False) and (r['flush_complete'] is True) and (all(r['children_reaped']))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'new_admission_closed': False, 'inflight_zero': 0, 'registration_removed': False, 'flush_complete': True, 'children_reaped': [True, True]}
check('valid operation', solve(r), True)
check('Shutdown still permits new work to enter', solve(dict(r, **{'new_admission_closed': True})), False)
check('Shutdown terminates in-flight requests prematurely', solve(dict(r, **{'inflight_zero': 1})), False)
check('Shutdown completes while routing still advertises the instance', solve(dict(r, **{'registration_removed': True})), False)
check('Shutdown discards buffered durable work', solve(dict(r, **{'flush_complete': False})), False)
check('Shutdown abandons child workers holding resources', solve(dict(r, **{'children_reaped': [True, False]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'new_admission_closed': True, 'inflight_zero': 1, 'registration_removed': True, 'flush_complete': False, 'children_reaped': [True, False]}
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 graceful drain operation is admitted even though shutdown abandons child workers holding resources.

Return a Boolean admission decision for finish draining a service instance. The record r must satisfy all of: r['new_admission_closed'] is False; r['inflight_zero'] == 0; r['registration_removed'] is False; r['flush_complete'] is True; all(r['children_reaped']). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access