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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enabled', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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)

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'disabled', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 enable-toggle event leaves a notification in enabled instead of enable-pending.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'disabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enable-error', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 save-success event leaves a notification in disabled instead of enabled.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enabled', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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)

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'disabled', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 save-failure event leaves a notification in enabled instead of enable-error.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disabled', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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)

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'enable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 retry-disable event leaves a notification in disabled instead of disable-pending.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enabled', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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)

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'disable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 retry-enable event leaves a notification in enabled instead of enable-pending.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'enabled', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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)

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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'enable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}
    data = json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})) if payload is None else {**json.loads(json.dumps({'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event.startswith('save-') and arg.get('request',data['request']) != data['request']: event = 'unknown-event'
            if event in ('disable-toggle','retry-disable'): data['desired'] = False
            if event in ('enable-toggle','retry-enable','toggle-back-on'): data['desired'] = True
            state = table.get((state, event), state)
            if state == 'disabled': data['channels'][data['channel']] = False
            if state == 'enabled': data['channels'][data['channel']] = True
            if state == 'enable-pending' and previous == 'cancel-disable-pending': data['request'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('enabled / disable-toggle', solve('enabled', [('disable-toggle', N)], N), 'disable-pending')
check('disable-pending / save-success', solve('disable-pending', [('save-success', N)], N), 'disabled')
check('disable-pending / save-failure', solve('disable-pending', [('save-failure', N)], N), 'disable-error')
check('disabled / enable-toggle', solve('disabled', [('enable-toggle', N)], N), 'enable-pending')
check('enable-pending / save-success', solve('enable-pending', [('save-success', N)], N), 'enabled')
check('enable-pending / save-failure', solve('enable-pending', [('save-failure', N)], N), 'enable-error')
check('disable-error / retry-disable', solve('disable-error', [('retry-disable', N)], N), 'disable-pending')
check('enable-error / retry-enable', solve('enable-error', [('retry-enable', N)], N), 'enable-pending')
check('disable-pending / toggle-back-on', solve('disable-pending', [('toggle-back-on', N)], N), 'cancel-disable-pending')
check('cancel-disable-pending / disable-save-completes', solve('cancel-disable-pending', [('disable-save-completes', N)], N), 'enable-pending')
check('older notification incarnation', solve('enabled', [('disable-toggle', N - 1)], N), 'enabled')
check('future notification incarnation', solve('enabled', [('disable-toggle', N + 1)], N), 'enabled')
check('unknown event is inert', solve('enabled', [('unknown-event', N)], N), 'enabled')
check('empty delivery batch', solve('enabled', [], N), 'enabled')
check('N stale deliveries before current delivery', solve('enabled', [('disable-toggle', N - 1)] * N + [('disable-toggle', N)], N), 'disable-pending')
check('trace disable-toggle then save-success', solve('enabled', [('disable-toggle', N), ('save-success', N)], N), 'disabled')
check('trace disable-toggle then save-failure', solve('enabled', [('disable-toggle', N), ('save-failure', N)], N), 'disable-error')
check('trace disable-toggle then toggle-back-on', solve('enabled', [('disable-toggle', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace save-success then enable-toggle', solve('disable-pending', [('save-success', N), ('enable-toggle', N)], N), 'enable-pending')
check('trace save-failure then retry-disable', solve('disable-pending', [('save-failure', N), ('retry-disable', N)], N), 'disable-pending')
check('trace enable-toggle then save-success', solve('disabled', [('enable-toggle', N), ('save-success', N)], N), 'enabled')
check('trace enable-toggle then save-failure', solve('disabled', [('enable-toggle', N), ('save-failure', N)], N), 'enable-error')
check('trace save-success then disable-toggle', solve('enable-pending', [('save-success', N), ('disable-toggle', N)], N), 'disable-pending')
check('trace save-failure then retry-enable', solve('enable-pending', [('save-failure', N), ('retry-enable', N)], N), 'enable-pending')
check('trace retry-disable then save-success', solve('disable-error', [('retry-disable', N), ('save-success', N)], N), 'disabled')
check('trace retry-disable then save-failure', solve('disable-error', [('retry-disable', N), ('save-failure', N)], N), 'disable-error')
check('trace retry-disable then toggle-back-on', solve('disable-error', [('retry-disable', N), ('toggle-back-on', N)], N), 'cancel-disable-pending')
check('trace retry-enable then save-success', solve('enable-error', [('retry-enable', N), ('save-success', N)], N), 'enabled')
check('trace retry-enable then save-failure', solve('enable-error', [('retry-enable', N), ('save-failure', N)], N), 'enable-error')
check('trace toggle-back-on then disable-save-completes', solve('disable-pending', [('toggle-back-on', N), ('disable-save-completes', N)], N), 'enable-pending')
check('trace disable-save-completes then save-success', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-success', N)], N), 'enabled')
check('trace disable-save-completes then save-failure', solve('cancel-disable-pending', [('disable-save-completes', N), ('save-failure', N)], N), 'enable-error')
check('domain state regression 1: channels', solve('disable-pending', [('save-success', N, {'request': 1})], N, {}, 'channels'), {'phase': 'disable-pending', 'value': {'mentions': True, 'security': True}})
check('domain state regression 2: channels', solve('disable-pending', [('save-success', N, {})], N, {}, 'channels'), {'phase': 'disabled', 'value': {'mentions': False, 'security': True}})
check('domain state regression 3: channels', solve('disable-pending', [('save-failure', N, {})], N, {}, 'channels'), {'phase': 'disable-error', 'value': {'mentions': True, 'security': True}})
check('domain state regression 4: desired', solve('disabled', [('enable-toggle', N, {})], N, {'channels': {'mentions': False, 'security': True}}, 'desired'), {'phase': 'enable-pending', 'value': True})
check('domain state regression 5: desired', solve('disable-pending', [('toggle-back-on', N, {})], N, {'desired': False}, 'desired'), {'phase': 'cancel-disable-pending', 'value': True})
check('domain state regression 6: request', solve('cancel-disable-pending', [('disable-save-completes', N, {})], N, {}, 'request'), {'phase': 'enable-pending', 'value': 3})
check('domain state regression 7: channels', solve('enabled', [('disable-toggle', N, {}), ('toggle-back-on', N, {}), ('disable-save-completes', N, {}), ('save-success', N, {'request': 3})], N, {}, 'channels'), {'phase': 'enabled', 'value': {'mentions': True, 'security': True}})
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 toggle-back-on event leaves a notification in enabled instead of cancel-disable-pending.

Notification channel preference feedback is a bounded visual-notification workflow with mutable policy state {'channel': 'mentions', 'channels': {'mentions': True, 'security': True}, 'request': 2, 'desired': True}. Its default transition relation is {('enabled', 'disable-toggle'): 'disable-pending', ('disable-pending', 'save-success'): 'disabled', ('disable-pending', 'save-failure'): 'disable-error', ('disabled', 'enable-toggle'): 'enable-pending', ('enable-pending', 'save-success'): 'enabled', ('enable-pending', 'save-failure'): 'enable-error', ('disable-error', 'retry-disable'): 'disable-pending', ('enable-error', 'retry-enable'): 'enable-pending', ('disable-pending', 'toggle-back-on'): 'cancel-disable-pending', ('cancel-disable-pending', 'disable-save-completes'): 'enable-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access