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

N = 1
observations = []
def solve(r):
    return (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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-failover-promote

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

N = 1
observations = []
def solve(r):
    return (r['primary_fenced'] is True) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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 failover promote operation is admitted even though promotion permits two writable primaries.

Return a Boolean admission decision for promote a standby after primary failure. The record r must satisfy all of: r['primary_fenced'] is True; r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]; r['standby_writable'] is True; r['promotion_epoch'][0] > r['promotion_epoch'][1]; r['routing_switched'][0] == r['routing_switched'][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_fenced'] is True) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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-failover-promote

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

N = 1
observations = []
def solve(r):
    return (r['primary_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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 failover promote operation is admitted even though standby lag exceeds the declared data-loss allowance.

Return a Boolean admission decision for promote a standby after primary failure. The record r must satisfy all of: r['primary_fenced'] is True; r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]; r['standby_writable'] is True; r['promotion_epoch'][0] > r['promotion_epoch'][1]; r['routing_switched'][0] == r['routing_switched'][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_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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-failover-promote

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

N = 1
observations = []
def solve(r):
    return (r['primary_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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 failover promote operation is admitted even though a read-only standby is advertised as write primary.

Return a Boolean admission decision for promote a standby after primary failure. The record r must satisfy all of: r['primary_fenced'] is True; r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]; r['standby_writable'] is True; r['promotion_epoch'][0] > r['promotion_epoch'][1]; r['routing_switched'][0] == r['routing_switched'][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_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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-failover-promote

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

N = 1
observations = []
def solve(r):
    return (r['primary_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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 failover promote operation is admitted even though a delayed promotion reverses a newer failover.

Return a Boolean admission decision for promote a standby after primary failure. The record r must satisfy all of: r['primary_fenced'] is True; r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]; r['standby_writable'] is True; r['promotion_epoch'][0] > r['promotion_epoch'][1]; r['routing_switched'][0] == r['routing_switched'][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_fenced'] is True) and (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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-failover-promote

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

N = 1
observations = []
def solve(r):
    return (r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]) and (r['standby_writable'] is True) and (r['promotion_epoch'][0] > r['promotion_epoch'][1]) and (r['routing_switched'][0] == r['routing_switched'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'primary_fenced': True, 'rpo_bound': [10, 9, 1], 'standby_writable': True, 'promotion_epoch': [6, 5], 'routing_switched': ['new', 'new']}
check('valid operation', solve(r), True)
check('Promotion permits two writable primaries', solve(dict(r, **{'primary_fenced': False})), False)
check('Standby lag exceeds the declared data-loss allowance', solve(dict(r, **{'rpo_bound': [10, 8, 1]})), False)
check('A read-only standby is advertised as write primary', solve(dict(r, **{'standby_writable': False})), False)
check('A delayed promotion reverses a newer failover', solve(dict(r, **{'promotion_epoch': [5, 5]})), False)
check('Promotion succeeds while writes still route to the old primary', solve(dict(r, **{'routing_switched': ['old', 'new']})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'primary_fenced': False, 'rpo_bound': [10, 8, 1], 'standby_writable': False, 'promotion_epoch': [5, 5], 'routing_switched': ['old', 'new']}
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 failover promote operation is admitted even though promotion succeeds while writes still route to the old primary.

Return a Boolean admission decision for promote a standby after primary failure. The record r must satisfy all of: r['primary_fenced'] is True; r['rpo_bound'][0]-r['rpo_bound'][1] <= r['rpo_bound'][2]; r['standby_writable'] is True; r['promotion_epoch'][0] > r['promotion_epoch'][1]; r['routing_switched'][0] == r['routing_switched'][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['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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-workflow-start

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

N = 1
observations = []
def solve(r):
    return (r['execution_unique'][0] not in r['execution_unique'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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 workflow start operation is admitted even though a duplicate start creates a second active execution.

Return a Boolean admission decision for start a durable workflow execution. The record r must satisfy all of: r['execution_unique'][0] not in r['execution_unique'][1]; r['version_available'][0] in r['version_available'][1]; set(r['input_required'][0]) <= set(r['input_required'][1]); r['parent_live'] == 'running'; r['start_deadline'][0] < r['start_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['execution_unique'][0] not in r['execution_unique'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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-workflow-start

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

N = 1
observations = []
def solve(r):
    return (r['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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 workflow start operation is admitted even though the workflow starts on an unavailable definition version.

Return a Boolean admission decision for start a durable workflow execution. The record r must satisfy all of: r['execution_unique'][0] not in r['execution_unique'][1]; r['version_available'][0] in r['version_available'][1]; set(r['input_required'][0]) <= set(r['input_required'][1]); r['parent_live'] == 'running'; r['start_deadline'][0] < r['start_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['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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-workflow-start

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

N = 1
observations = []
def solve(r):
    return (r['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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 workflow start operation is admitted even though a workflow starts without its required input fields.

Return a Boolean admission decision for start a durable workflow execution. The record r must satisfy all of: r['execution_unique'][0] not in r['execution_unique'][1]; r['version_available'][0] in r['version_available'][1]; set(r['input_required'][0]) <= set(r['input_required'][1]); r['parent_live'] == 'running'; r['start_deadline'][0] < r['start_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['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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-workflow-start

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

N = 1
observations = []
def solve(r):
    return (r['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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 workflow start operation is admitted even though a child starts after its parent was cancelled.

Return a Boolean admission decision for start a durable workflow execution. The record r must satisfy all of: r['execution_unique'][0] not in r['execution_unique'][1]; r['version_available'][0] in r['version_available'][1]; set(r['input_required'][0]) <= set(r['input_required'][1]); r['parent_live'] == 'running'; r['start_deadline'][0] < r['start_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['execution_unique'][0] not in r['execution_unique'][1]) and (r['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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-workflow-start

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

N = 1
observations = []
def solve(r):
    return (r['version_available'][0] in r['version_available'][1]) and (set(r['input_required'][0]) <= set(r['input_required'][1])) and (r['parent_live'] == 'running') and (r['start_deadline'][0] < r['start_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'execution_unique': ['e2', ['e1']], 'version_available': [3, [2, 3]], 'input_required': [['order'], ['order', 'note']], 'parent_live': 'running', 'start_deadline': [10, 11]}
check('valid operation', solve(r), True)
check('A duplicate start creates a second active execution', solve(dict(r, **{'execution_unique': ['e1', ['e1']]})), False)
check('The workflow starts on an unavailable definition version', solve(dict(r, **{'version_available': [4, [2, 3]]})), False)
check('A workflow starts without its required input fields', solve(dict(r, **{'input_required': [['order'], ['note']]})), False)
check('A child starts after its parent was cancelled', solve(dict(r, **{'parent_live': 'cancelled'})), False)
check('A deferred start begins beyond the workflow start deadline', solve(dict(r, **{'start_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 = {'execution_unique': ['e1', ['e1']], 'version_available': [4, [2, 3]], 'input_required': [['order'], ['note']], 'parent_live': 'cancelled', 'start_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 workflow start operation is admitted even though a deferred start begins beyond the workflow start deadline.

Return a Boolean admission decision for start a durable workflow execution. The record r must satisfy all of: r['execution_unique'][0] not in r['execution_unique'][1]; r['version_available'][0] in r['version_available'][1]; set(r['input_required'][0]) <= set(r['input_required'][1]); r['parent_live'] == 'running'; r['start_deadline'][0] < r['start_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['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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-activity-completion

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

N = 1
observations = []
def solve(r):
    return (r['run_identity'][0] == r['run_identity'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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 activity completion operation is admitted even though a completion from a previous workflow run is accepted.

Return a Boolean admission decision for accept a durable activity completion. The record r must satisfy all of: r['run_identity'][0] == r['run_identity'][1]; r['activity_attempt'][0] == r['activity_attempt'][1]; r['activity_scheduled'][0] in r['activity_scheduled'][1]; set(r['result_schema'][0]) <= set(r['result_schema'][1]); r['completion_pending'] == 'pending'. 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['run_identity'][0] == r['run_identity'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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-activity-completion

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

N = 1
observations = []
def solve(r):
    return (r['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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 activity completion operation is admitted even though a timed-out activity attempt overwrites its successor.

Return a Boolean admission decision for accept a durable activity completion. The record r must satisfy all of: r['run_identity'][0] == r['run_identity'][1]; r['activity_attempt'][0] == r['activity_attempt'][1]; r['activity_scheduled'][0] in r['activity_scheduled'][1]; set(r['result_schema'][0]) <= set(r['result_schema'][1]); r['completion_pending'] == 'pending'. 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['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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-activity-completion

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

N = 1
observations = []
def solve(r):
    return (r['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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 activity completion operation is admitted even though an unsolicited completion invents an unscheduled activity.

Return a Boolean admission decision for accept a durable activity completion. The record r must satisfy all of: r['run_identity'][0] == r['run_identity'][1]; r['activity_attempt'][0] == r['activity_attempt'][1]; r['activity_scheduled'][0] in r['activity_scheduled'][1]; set(r['result_schema'][0]) <= set(r['result_schema'][1]); r['completion_pending'] == 'pending'. 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['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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-activity-completion

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

N = 1
observations = []
def solve(r):
    return (r['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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 activity completion operation is admitted even though an incompatible activity result enters durable history.

Return a Boolean admission decision for accept a durable activity completion. The record r must satisfy all of: r['run_identity'][0] == r['run_identity'][1]; r['activity_attempt'][0] == r['activity_attempt'][1]; r['activity_scheduled'][0] in r['activity_scheduled'][1]; set(r['result_schema'][0]) <= set(r['result_schema'][1]); r['completion_pending'] == 'pending'. 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['run_identity'][0] == r['run_identity'][1]) and (r['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1]))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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-activity-completion

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

N = 1
observations = []
def solve(r):
    return (r['activity_attempt'][0] == r['activity_attempt'][1]) and (r['activity_scheduled'][0] in r['activity_scheduled'][1]) and (set(r['result_schema'][0]) <= set(r['result_schema'][1])) and (r['completion_pending'] == 'pending')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'run_identity': ['r2', 'r2'], 'activity_attempt': [3, 3], 'activity_scheduled': ['a', ['a', 'b']], 'result_schema': [['receipt'], ['receipt']], 'completion_pending': 'pending'}
check('valid operation', solve(r), True)
check('A completion from a previous workflow run is accepted', solve(dict(r, **{'run_identity': ['r1', 'r2']})), False)
check('A timed-out activity attempt overwrites its successor', solve(dict(r, **{'activity_attempt': [2, 3]})), False)
check('An unsolicited completion invents an unscheduled activity', solve(dict(r, **{'activity_scheduled': ['c', ['a', 'b']]})), False)
check('An incompatible activity result enters durable history', solve(dict(r, **{'result_schema': [['receipt'], ['error']]})), False)
check('A duplicate activity completion schedules downstream work twice', solve(dict(r, **{'completion_pending': 'completed'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'run_identity': ['r1', 'r2'], 'activity_attempt': [2, 3], 'activity_scheduled': ['c', ['a', 'b']], 'result_schema': [['receipt'], ['error']], 'completion_pending': 'completed'}
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 activity completion operation is admitted even though a duplicate activity completion schedules downstream work twice.

Return a Boolean admission decision for accept a durable activity completion. The record r must satisfy all of: r['run_identity'][0] == r['run_identity'][1]; r['activity_attempt'][0] == r['activity_attempt'][1]; r['activity_scheduled'][0] in r['activity_scheduled'][1]; set(r['result_schema'][0]) <= set(r['result_schema'][1]); r['completion_pending'] == 'pending'. 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['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-timer

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

N = 1
observations = []
def solve(r):
    return (r['fire_time'][0] >= r['fire_time'][1]) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow timer operation is admitted even though a timer fires before its scheduled timestamp.

Return a Boolean admission decision for fire a durable workflow timer. The record r must satisfy all of: r['fire_time'][0] >= r['fire_time'][1]; r['timer_not_cancelled'] is False; r['timer_generation'][0] == r['timer_generation'][1]; r['workflow_not_terminal'] not in ['completed','failed','cancelled']; r['fire_once'] is False. 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['fire_time'][0] >= r['fire_time'][1]) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-timer

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

N = 1
observations = []
def solve(r):
    return (r['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow timer operation is admitted even though a cancelled timer resumes workflow execution.

Return a Boolean admission decision for fire a durable workflow timer. The record r must satisfy all of: r['fire_time'][0] >= r['fire_time'][1]; r['timer_not_cancelled'] is False; r['timer_generation'][0] == r['timer_generation'][1]; r['workflow_not_terminal'] not in ['completed','failed','cancelled']; r['fire_once'] is False. 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['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-timer

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

N = 1
observations = []
def solve(r):
    return (r['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow timer operation is admitted even though a stale timer wakeup fires a replacement timer.

Return a Boolean admission decision for fire a durable workflow timer. The record r must satisfy all of: r['fire_time'][0] >= r['fire_time'][1]; r['timer_not_cancelled'] is False; r['timer_generation'][0] == r['timer_generation'][1]; r['workflow_not_terminal'] not in ['completed','failed','cancelled']; r['fire_once'] is False. 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['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-timer

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

N = 1
observations = []
def solve(r):
    return (r['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow timer operation is admitted even though a timer reactivates a completed workflow.

Return a Boolean admission decision for fire a durable workflow timer. The record r must satisfy all of: r['fire_time'][0] >= r['fire_time'][1]; r['timer_not_cancelled'] is False; r['timer_generation'][0] == r['timer_generation'][1]; r['workflow_not_terminal'] not in ['completed','failed','cancelled']; r['fire_once'] is False. 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['fire_time'][0] >= r['fire_time'][1]) and (r['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled'])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-timer

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

N = 1
observations = []
def solve(r):
    return (r['timer_not_cancelled'] is False) and (r['timer_generation'][0] == r['timer_generation'][1]) and (r['workflow_not_terminal'] not in ['completed','failed','cancelled']) and (r['fire_once'] is False)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'fire_time': [20, 20], 'timer_not_cancelled': False, 'timer_generation': [2, 2], 'workflow_not_terminal': 'waiting', 'fire_once': False}
check('valid operation', solve(r), True)
check('A timer fires before its scheduled timestamp', solve(dict(r, **{'fire_time': [19, 20]})), False)
check('A cancelled timer resumes workflow execution', solve(dict(r, **{'timer_not_cancelled': True})), False)
check('A stale timer wakeup fires a replacement timer', solve(dict(r, **{'timer_generation': [1, 2]})), False)
check('A timer reactivates a completed workflow', solve(dict(r, **{'workflow_not_terminal': 'completed'})), False)
check('A retried timer delivery produces another history event', solve(dict(r, **{'fire_once': True})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'fire_time': [19, 20], 'timer_not_cancelled': True, 'timer_generation': [1, 2], 'workflow_not_terminal': 'completed', 'fire_once': True}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow timer operation is admitted even though a retried timer delivery produces another history event.

Return a Boolean admission decision for fire a durable workflow timer. The record r must satisfy all of: r['fire_time'][0] >= r['fire_time'][1]; r['timer_not_cancelled'] is False; r['timer_generation'][0] == r['timer_generation'][1]; r['workflow_not_terminal'] not in ['completed','failed','cancelled']; r['fire_once'] is False. 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['recipient_run'][0] == r['recipient_run'][1]) and (r['signal_declared'][0] in r['signal_declared'][1]) and (r['mailbox_capacity'][0] < r['mailbox_capacity'][1]) and (r['sequence_dependency'][0] <= r['sequence_dependency'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'signal_identity': ['s2', ['s1']], 'recipient_run': ['run2', 'run2'], 'signal_declared': ['approve', ['approve', 'reject']], 'mailbox_capacity': [9, 10], 'sequence_dependency': [4, 4]}
check('valid operation', solve(r), True)
check('Duplicate signal delivery applies the external command twice', solve(dict(r, **{'signal_identity': ['s1', ['s1']]})), False)
check('A delayed signal reaches a different run under the same workflow name', solve(dict(r, **{'recipient_run': ['run1', 'run2']})), False)
check('An unknown signal type bypasses workflow dispatch', solve(dict(r, **{'signal_declared': ['delete', ['approve', 'reject']]})), False)
check('External signals exhaust the workflow pending-event budget', solve(dict(r, **{'mailbox_capacity': [10, 10]})), False)
check('A signal is applied before its required earlier command', solve(dict(r, **{'sequence_dependency': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'signal_identity': ['s1', ['s1']], 'recipient_run': ['run1', 'run2'], 'signal_declared': ['delete', ['approve', 'reject']], 'mailbox_capacity': [10, 10], 'sequence_dependency': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-signal

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

N = 1
observations = []
def solve(r):
    return (r['signal_identity'][0] not in r['signal_identity'][1]) and (r['signal_declared'][0] in r['signal_declared'][1]) and (r['mailbox_capacity'][0] < r['mailbox_capacity'][1]) and (r['sequence_dependency'][0] <= r['sequence_dependency'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'signal_identity': ['s2', ['s1']], 'recipient_run': ['run2', 'run2'], 'signal_declared': ['approve', ['approve', 'reject']], 'mailbox_capacity': [9, 10], 'sequence_dependency': [4, 4]}
check('valid operation', solve(r), True)
check('Duplicate signal delivery applies the external command twice', solve(dict(r, **{'signal_identity': ['s1', ['s1']]})), False)
check('A delayed signal reaches a different run under the same workflow name', solve(dict(r, **{'recipient_run': ['run1', 'run2']})), False)
check('An unknown signal type bypasses workflow dispatch', solve(dict(r, **{'signal_declared': ['delete', ['approve', 'reject']]})), False)
check('External signals exhaust the workflow pending-event budget', solve(dict(r, **{'mailbox_capacity': [10, 10]})), False)
check('A signal is applied before its required earlier command', solve(dict(r, **{'sequence_dependency': [5, 4]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'signal_identity': ['s1', ['s1']], 'recipient_run': ['run1', 'run2'], 'signal_declared': ['delete', ['approve', 'reject']], 'mailbox_capacity': [10, 10], 'sequence_dependency': [5, 4]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow signal operation is admitted even though duplicate signal delivery applies the external command twice.

Return a Boolean admission decision for deliver a signal to a durable workflow. The record r must satisfy all of: r['signal_identity'][0] not in r['signal_identity'][1]; r['recipient_run'][0] == r['recipient_run'][1]; r['signal_declared'][0] in r['signal_declared'][1]; r['mailbox_capacity'][0] < r['mailbox_capacity'][1]; r['sequence_dependency'][0] <= r['sequence_dependency'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access