broken_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)

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['recipient_run'][0] == r['recipient_run'][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 a delayed signal reaches a different run under the same workflow name.

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


broken_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['recipient_run'][0] == r['recipient_run'][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)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

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['recipient_run'][0] == r['recipient_run'][1]) and (r['signal_declared'][0] in r['signal_declared'][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 an unknown signal type bypasses workflow dispatch.

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


broken_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['recipient_run'][0] == r['recipient_run'][1]) and (r['signal_declared'][0] in r['signal_declared'][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['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])
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 external signals exhaust the workflow pending-event budget.

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


broken_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['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])
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['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)

prompt:
The workflow signal operation is admitted even though a signal is applied before its required earlier command.

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


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

N = 1
observations = []
def solve(r):
    return (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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-saga-compensation

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

N = 1
observations = []
def solve(r):
    return (r['forward_committed'] is True) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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 saga compensation operation is admitted even though compensation reverses a forward action that never committed.

Return a Boolean admission decision for run a saga compensation action. The record r must satisfy all of: r['forward_committed'] is True; all(r['reverse_dependency']); r['compensation_identity'][0] not in r['compensation_identity'][1]; r['resource_version'][0] == r['resource_version'][1]; r['compensation_registered'] is not None. 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['forward_committed'] is True) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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-saga-compensation

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

N = 1
observations = []
def solve(r):
    return (r['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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 saga compensation operation is admitted even though an upstream step is compensated while dependents remain active.

Return a Boolean admission decision for run a saga compensation action. The record r must satisfy all of: r['forward_committed'] is True; all(r['reverse_dependency']); r['compensation_identity'][0] not in r['compensation_identity'][1]; r['resource_version'][0] == r['resource_version'][1]; r['compensation_registered'] is not None. 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['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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-saga-compensation

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

N = 1
observations = []
def solve(r):
    return (r['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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 saga compensation operation is admitted even though retried compensation refunds the same side effect twice.

Return a Boolean admission decision for run a saga compensation action. The record r must satisfy all of: r['forward_committed'] is True; all(r['reverse_dependency']); r['compensation_identity'][0] not in r['compensation_identity'][1]; r['resource_version'][0] == r['resource_version'][1]; r['compensation_registered'] is not None. 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['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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-saga-compensation

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

N = 1
observations = []
def solve(r):
    return (r['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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 saga compensation operation is admitted even though compensation overwrites a resource changed by another workflow.

Return a Boolean admission decision for run a saga compensation action. The record r must satisfy all of: r['forward_committed'] is True; all(r['reverse_dependency']); r['compensation_identity'][0] not in r['compensation_identity'][1]; r['resource_version'][0] == r['resource_version'][1]; r['compensation_registered'] is not None. 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['forward_committed'] is True) and (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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-saga-compensation

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

N = 1
observations = []
def solve(r):
    return (all(r['reverse_dependency'])) and (r['compensation_identity'][0] not in r['compensation_identity'][1]) and (r['resource_version'][0] == r['resource_version'][1]) and (r['compensation_registered'] is not None)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'forward_committed': True, 'reverse_dependency': [True, True], 'compensation_identity': ['c2', ['c1']], 'resource_version': [7, 7], 'compensation_registered': 'release'}
check('valid operation', solve(r), True)
check('Compensation reverses a forward action that never committed', solve(dict(r, **{'forward_committed': False})), False)
check('An upstream step is compensated while dependents remain active', solve(dict(r, **{'reverse_dependency': [True, False]})), False)
check('Retried compensation refunds the same side effect twice', solve(dict(r, **{'compensation_identity': ['c1', ['c1']]})), False)
check('Compensation overwrites a resource changed by another workflow', solve(dict(r, **{'resource_version': [6, 7]})), False)
check('A committed step has no recorded inverse operation', solve(dict(r, **{'compensation_registered': None})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'forward_committed': False, 'reverse_dependency': [True, False], 'compensation_identity': ['c1', ['c1']], 'resource_version': [6, 7], 'compensation_registered': None}
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 saga compensation operation is admitted even though a committed step has no recorded inverse operation.

Return a Boolean admission decision for run a saga compensation action. The record r must satisfy all of: r['forward_committed'] is True; all(r['reverse_dependency']); r['compensation_identity'][0] not in r['compensation_identity'][1]; r['resource_version'][0] == r['resource_version'][1]; r['compensation_registered'] is not None. 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['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-replay

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

N = 1
observations = []
def solve(r):
    return (r['command_type'][0] == r['command_type'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow replay operation is admitted even though replayed code emits a different command kind than history.

Return a Boolean admission decision for replay durable workflow history. The record r must satisfy all of: r['command_type'][0] == r['command_type'][1]; r['command_order'][0] == r['command_order'][1]; r['version_marker'][0] in r['version_marker'][1]; r['random_recorded'] is True; r['history_contiguous'] == list(range(len(r['history_contiguous']))). 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['command_type'][0] == r['command_type'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-replay

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

N = 1
observations = []
def solve(r):
    return (r['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow replay operation is admitted even though replayed commands reverse the recorded causal order.

Return a Boolean admission decision for replay durable workflow history. The record r must satisfy all of: r['command_type'][0] == r['command_type'][1]; r['command_order'][0] == r['command_order'][1]; r['version_marker'][0] in r['version_marker'][1]; r['random_recorded'] is True; r['history_contiguous'] == list(range(len(r['history_contiguous']))). 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['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-replay

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

N = 1
observations = []
def solve(r):
    return (r['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow replay operation is admitted even though code executes a branch not selected by the recorded marker.

Return a Boolean admission decision for replay durable workflow history. The record r must satisfy all of: r['command_type'][0] == r['command_type'][1]; r['command_order'][0] == r['command_order'][1]; r['version_marker'][0] in r['version_marker'][1]; r['random_recorded'] is True; r['history_contiguous'] == list(range(len(r['history_contiguous']))). 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['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-replay

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

N = 1
observations = []
def solve(r):
    return (r['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow replay operation is admitted even though replay consumes an unrecorded nondeterministic random value.

Return a Boolean admission decision for replay durable workflow history. The record r must satisfy all of: r['command_type'][0] == r['command_type'][1]; r['command_order'][0] == r['command_order'][1]; r['version_marker'][0] in r['version_marker'][1]; r['random_recorded'] is True; r['history_contiguous'] == list(range(len(r['history_contiguous']))). 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['command_type'][0] == r['command_type'][1]) and (r['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True)
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-replay

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

N = 1
observations = []
def solve(r):
    return (r['command_order'][0] == r['command_order'][1]) and (r['version_marker'][0] in r['version_marker'][1]) and (r['random_recorded'] is True) and (r['history_contiguous'] == list(range(len(r['history_contiguous']))))
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'command_type': ['timer', 'timer'], 'command_order': [['a', 'b'], ['a', 'b']], 'version_marker': [2, [1, 2]], 'random_recorded': True, 'history_contiguous': [0, 1, 2]}
check('valid operation', solve(r), True)
check('Replayed code emits a different command kind than history', solve(dict(r, **{'command_type': ['activity', 'timer']})), False)
check('Replayed commands reverse the recorded causal order', solve(dict(r, **{'command_order': [['b', 'a'], ['a', 'b']]})), False)
check('Code executes a branch not selected by the recorded marker', solve(dict(r, **{'version_marker': [3, [1, 2]]})), False)
check('Replay consumes an unrecorded nondeterministic random value', solve(dict(r, **{'random_recorded': False})), False)
check('Replay silently skips a missing history event', solve(dict(r, **{'history_contiguous': [0, 2, 3]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'command_type': ['activity', 'timer'], 'command_order': [['b', 'a'], ['a', 'b']], 'version_marker': [3, [1, 2]], 'random_recorded': False, 'history_contiguous': [0, 2, 3]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow replay operation is admitted even though replay silently skips a missing history event.

Return a Boolean admission decision for replay durable workflow history. The record r must satisfy all of: r['command_type'][0] == r['command_type'][1]; r['command_order'][0] == r['command_order'][1]; r['version_marker'][0] in r['version_marker'][1]; r['random_recorded'] is True; r['history_contiguous'] == list(range(len(r['history_contiguous']))). Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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-continue

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

N = 1
observations = []
def solve(r):
    return (r['pending_activities'] == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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 continue operation is admitted even though continuation loses unresolved activity results.

Return a Boolean admission decision for continue a workflow as a new run. The record r must satisfy all of: r['pending_activities'] == 0; len(r['signal_drain']) == 0; r['state_serializable'] in ['json','bytes']; r['new_run_unique'][0] != r['new_run_unique'][1]; r['predecessor_closed'] == 'continued'. 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['pending_activities'] == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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-continue

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

N = 1
observations = []
def solve(r):
    return (r['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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 continue operation is admitted even though continuation discards accepted but unconsumed signals.

Return a Boolean admission decision for continue a workflow as a new run. The record r must satisfy all of: r['pending_activities'] == 0; len(r['signal_drain']) == 0; r['state_serializable'] in ['json','bytes']; r['new_run_unique'][0] != r['new_run_unique'][1]; r['predecessor_closed'] == 'continued'. 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['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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-continue

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

N = 1
observations = []
def solve(r):
    return (r['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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 continue operation is admitted even though continuation stores a runtime handle instead of durable state.

Return a Boolean admission decision for continue a workflow as a new run. The record r must satisfy all of: r['pending_activities'] == 0; len(r['signal_drain']) == 0; r['state_serializable'] in ['json','bytes']; r['new_run_unique'][0] != r['new_run_unique'][1]; r['predecessor_closed'] == 'continued'. 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['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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-continue

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

N = 1
observations = []
def solve(r):
    return (r['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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 continue operation is admitted even though continuation reuses the predecessor run identifier.

Return a Boolean admission decision for continue a workflow as a new run. The record r must satisfy all of: r['pending_activities'] == 0; len(r['signal_drain']) == 0; r['state_serializable'] in ['json','bytes']; r['new_run_unique'][0] != r['new_run_unique'][1]; r['predecessor_closed'] == 'continued'. 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['pending_activities'] == 0) and (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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-continue

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

N = 1
observations = []
def solve(r):
    return (len(r['signal_drain']) == 0) and (r['state_serializable'] in ['json','bytes']) and (r['new_run_unique'][0] != r['new_run_unique'][1]) and (r['predecessor_closed'] == 'continued')
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'pending_activities': 0, 'signal_drain': [], 'state_serializable': 'json', 'new_run_unique': ['r2', 'r1'], 'predecessor_closed': 'continued'}
check('valid operation', solve(r), True)
check('Continuation loses unresolved activity results', solve(dict(r, **{'pending_activities': 1})), False)
check('Continuation discards accepted but unconsumed signals', solve(dict(r, **{'signal_drain': ['approve']})), False)
check('Continuation stores a runtime handle instead of durable state', solve(dict(r, **{'state_serializable': 'socket'})), False)
check('Continuation reuses the predecessor run identifier', solve(dict(r, **{'new_run_unique': ['r1', 'r1']})), False)
check('Both predecessor and successor remain runnable', solve(dict(r, **{'predecessor_closed': 'running'})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'pending_activities': 1, 'signal_drain': ['approve'], 'state_serializable': 'socket', 'new_run_unique': ['r1', 'r1'], 'predecessor_closed': 'running'}
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 continue operation is admitted even though both predecessor and successor remain runnable.

Return a Boolean admission decision for continue a workflow as a new run. The record r must satisfy all of: r['pending_activities'] == 0; len(r['signal_drain']) == 0; r['state_serializable'] in ['json','bytes']; r['new_run_unique'][0] != r['new_run_unique'][1]; r['predecessor_closed'] == 'continued'. 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_cleanup'] is True) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-cancel

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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow cancel operation is admitted even though cancellation leaves dependent child workflows executing.

Return a Boolean admission decision for complete workflow cancellation. The record r must satisfy all of: all(r['child_propagation']); r['activity_cleanup'] is True; r['terminal_stability'] not in ['completed','failed']; r['request_generation'][0] == r['request_generation'][1]; r['cleanup_deadline'][0] <= r['cleanup_deadline'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access


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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['terminal_stability'] not in ['completed','failed']) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

evaluation_group: xd-workflow-cancel

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

N = 1
observations = []
def solve(r):
    return (all(r['child_propagation'])) and (r['activity_cleanup'] is True) and (r['request_generation'][0] == r['request_generation'][1]) and (r['cleanup_deadline'][0] <= r['cleanup_deadline'][1])
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
r = {'child_propagation': [True, True], 'activity_cleanup': True, 'terminal_stability': 'running', 'request_generation': ['r3', 'r3'], 'cleanup_deadline': [10, 10]}
check('valid operation', solve(r), True)
check('Cancellation leaves dependent child workflows executing', solve(dict(r, **{'child_propagation': [True, False]})), False)
check('Cancellation completes before required activity cleanup', solve(dict(r, **{'activity_cleanup': False})), False)
check('Cancellation rewrites an already successful terminal result', solve(dict(r, **{'terminal_stability': 'completed'})), False)
check('An old cancellation request terminates a newer run', solve(dict(r, **{'request_generation': ['r2', 'r3']})), False)
check('Cancellation waits beyond its explicit cleanup deadline', solve(dict(r, **{'cleanup_deadline': [11, 10]})), False)
check('unrelated tracing metadata', solve(dict(r, trace='run-'+str(N))), True)
check('repeat validation is pure', solve(r), True)
invalid = {'child_propagation': [True, False], 'activity_cleanup': False, 'terminal_stability': 'completed', 'request_generation': ['r2', 'r3'], 'cleanup_deadline': [11, 10]}
keys = list(invalid)
pair = {keys[N % len(keys)]: invalid[keys[N % len(keys)]], keys[(N+1) % len(keys)]: invalid[keys[(N+1) % len(keys)]]}
check('two independent violations in variant', solve(dict(r, **pair)), False)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The workflow cancel operation is admitted even though cancellation completes before required activity cleanup.

Return a Boolean admission decision for complete workflow cancellation. The record r must satisfy all of: all(r['child_propagation']); r['activity_cleanup'] is True; r['terminal_stability'] not in ['completed','failed']; r['request_generation'][0] == r['request_generation'][1]; r['cleanup_deadline'][0] <= r['cleanup_deadline'][1]. Extra tracing fields are ignored; validation does not mutate the record.

split: open-access