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 = {('received', 'mount-request'): 'seen', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'ack-pending', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 mount-request event leaves a notification in seen instead of mounting.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'seen', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'received', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 paint-confirmed event leaves a notification in seen instead of ack-pending.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'seen', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'ack-pending', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 mount-failed event leaves a notification in seen instead of undisplayed.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'received', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'undisplayed', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 ack-success event leaves a notification in received instead of seen.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'seen', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'mounting', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 ack-failure event leaves a notification in seen instead of ack-retry.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'mounting', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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 = {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'seen', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}
    data = json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})) if payload is None else {**json.loads(json.dumps({'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0})), **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 in ('paint-confirmed','mount-failed') and arg.get('mount',data['mount']) != data['mount']: event = 'unknown-event'
            if event == 'paint-confirmed': data['painted'] = True
            if event == 'retry-ack': data['ack_attempts'] += 1
            if event == 'ack-success' and 'require_paint' in arg and not data['painted']: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'seen': data['acknowledged'] = True
            if state == 'undisplayed': data['painted'] = False
    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('received / mount-request', solve('received', [('mount-request', N)], N), 'mounting')
check('mounting / paint-confirmed', solve('mounting', [('paint-confirmed', N)], N), 'ack-pending')
check('mounting / mount-failed', solve('mounting', [('mount-failed', N)], N), 'undisplayed')
check('ack-pending / ack-success', solve('ack-pending', [('ack-success', N)], N), 'seen')
check('ack-pending / ack-failure', solve('ack-pending', [('ack-failure', N)], N), 'ack-retry')
check('ack-retry / retry-ack', solve('ack-retry', [('retry-ack', N)], N), 'ack-pending')
check('undisplayed / retry-mount', solve('undisplayed', [('retry-mount', N)], N), 'mounting')
check('seen / delivery-replay', solve('seen', [('delivery-replay', N)], N), 'seen')
check('mounting / route-unmount', solve('mounting', [('route-unmount', N)], N), 'undisplayed')
check('ack-retry / surface-unmount', solve('ack-retry', [('surface-unmount', N)], N), 'ack-retry')
check('older notification incarnation', solve('received', [('mount-request', N - 1)], N), 'received')
check('future notification incarnation', solve('received', [('mount-request', N + 1)], N), 'received')
check('unknown event is inert', solve('received', [('unknown-event', N)], N), 'received')
check('empty delivery batch', solve('received', [], N), 'received')
check('N stale deliveries before current delivery', solve('received', [('mount-request', N - 1)] * N + [('mount-request', N)], N), 'mounting')
check('trace mount-request then paint-confirmed', solve('received', [('mount-request', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace mount-request then mount-failed', solve('received', [('mount-request', N), ('mount-failed', N)], N), 'undisplayed')
check('trace mount-request then route-unmount', solve('received', [('mount-request', N), ('route-unmount', N)], N), 'undisplayed')
check('trace paint-confirmed then ack-success', solve('mounting', [('paint-confirmed', N), ('ack-success', N)], N), 'seen')
check('trace paint-confirmed then ack-failure', solve('mounting', [('paint-confirmed', N), ('ack-failure', N)], N), 'ack-retry')
check('trace mount-failed then retry-mount', solve('mounting', [('mount-failed', N), ('retry-mount', N)], N), 'mounting')
check('trace ack-success then delivery-replay', solve('ack-pending', [('ack-success', N), ('delivery-replay', N)], N), 'seen')
check('trace ack-failure then retry-ack', solve('ack-pending', [('ack-failure', N), ('retry-ack', N)], N), 'ack-pending')
check('trace ack-failure then surface-unmount', solve('ack-pending', [('ack-failure', N), ('surface-unmount', N)], N), 'ack-retry')
check('trace retry-ack then ack-success', solve('ack-retry', [('retry-ack', N), ('ack-success', N)], N), 'seen')
check('trace retry-ack then ack-failure', solve('ack-retry', [('retry-ack', N), ('ack-failure', N)], N), 'ack-retry')
check('trace retry-mount then paint-confirmed', solve('undisplayed', [('retry-mount', N), ('paint-confirmed', N)], N), 'ack-pending')
check('trace retry-mount then mount-failed', solve('undisplayed', [('retry-mount', N), ('mount-failed', N)], N), 'undisplayed')
check('trace retry-mount then route-unmount', solve('undisplayed', [('retry-mount', N), ('route-unmount', N)], N), 'undisplayed')
check('trace delivery-replay then delivery-replay', solve('seen', [('delivery-replay', N), ('delivery-replay', N)], N), 'seen')
check('trace route-unmount then retry-mount', solve('mounting', [('route-unmount', N), ('retry-mount', N)], N), 'mounting')
check('trace surface-unmount then retry-ack', solve('ack-retry', [('surface-unmount', N), ('retry-ack', N)], N), 'ack-pending')
check('trace surface-unmount then surface-unmount', solve('ack-retry', [('surface-unmount', N), ('surface-unmount', N)], N), 'ack-retry')
check('domain state regression 1: painted', solve('mounting', [('paint-confirmed', N, {'mount': 3})], N, {}, 'painted'), {'phase': 'mounting', 'value': False})
check('domain state regression 2: painted', solve('mounting', [('paint-confirmed', N, {'mount': 4})], N, {}, 'painted'), {'phase': 'ack-pending', 'value': True})
check('domain state regression 3: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'ack-pending', 'value': False})
check('domain state regression 4: acknowledged', solve('ack-pending', [('ack-success', N, {'require_paint': True})], N, {'painted': True}, 'acknowledged'), {'phase': 'seen', 'value': True})
check('domain state regression 5: ack_attempts', solve('ack-retry', [('retry-ack', N, {})], N, {'painted': True, 'ack_attempts': 2}, 'ack_attempts'), {'phase': 'ack-pending', 'value': 3})
check('domain state regression 6: painted', solve('mounting', [('route-unmount', N, {})], N, {'painted': True}, 'painted'), {'phase': 'undisplayed', 'value': False})
check('domain state regression 7: acknowledged', solve('mounting', [('paint-confirmed', N, {}), ('ack-failure', N, {}), ('surface-unmount', N, {}), ('retry-ack', N, {}), ('ack-success', N, {'require_paint': True})], N, {}, 'acknowledged'), {'phase': 'seen', 'value': 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-ack event leaves a notification in mounting instead of ack-pending.

Visual notification delivery acknowledgement is a bounded visual-notification workflow with mutable policy state {'mount': 4, 'painted': False, 'acknowledged': False, 'ack_attempts': 0}. Its default transition relation is {('received', 'mount-request'): 'mounting', ('mounting', 'paint-confirmed'): 'ack-pending', ('mounting', 'mount-failed'): 'undisplayed', ('ack-pending', 'ack-success'): 'seen', ('ack-pending', 'ack-failure'): 'ack-retry', ('ack-retry', 'retry-ack'): 'ack-pending', ('undisplayed', 'retry-mount'): 'mounting', ('seen', 'delivery-replay'): 'seen', ('mounting', 'route-unmount'): 'undisplayed', ('ack-retry', 'surface-unmount'): 'ack-retry'}; 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