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 = {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'live', ('reconciling-preview', 'snapshot-current'): 'live-preview'}
    data = json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})) if payload is None else {**json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})), **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 == 'snapshot-current':
                if arg.get('cursor',data['cursor']) < data['cursor']: event = 'unknown-event'
                else:
                    data['cursor'] = arg.get('cursor',data['cursor'])
                    data['cached'] = arg.get('items',data['cached'])[:]
            if event == 'connection-lost': data['fresh'] = False
            state = table.get((state, event), state)
            if state in ('live','live-preview'): data['fresh'] = True
            if state == 'live-preview' and data['preview'] not in data['cached']: data['preview'] = None
            if state == 'empty': data['cached'] = []
    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('live / connection-lost', solve('live', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / connection-restored', solve('offline-cached', [('connection-restored', N)], N), 'reconciling')
check('reconciling / snapshot-current', solve('reconciling', [('snapshot-current', N)], N), 'live')
check('reconciling / snapshot-failed', solve('reconciling', [('snapshot-failed', N)], N), 'offline-error')
check('offline-error / retry-refresh', solve('offline-error', [('retry-refresh', N)], N), 'reconciling')
check('reconciling / connection-lost', solve('reconciling', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / open-cached-item', solve('offline-cached', [('open-cached-item', N)], N), 'offline-preview')
check('offline-preview / close-preview', solve('offline-preview', [('close-preview', N)], N), 'offline-cached')
check('offline-preview / connection-restored', solve('offline-preview', [('connection-restored', N)], N), 'reconciling-preview')
check('reconciling-preview / snapshot-current', solve('reconciling-preview', [('snapshot-current', N)], N), 'live-preview')
check('older notification incarnation', solve('live', [('connection-lost', N - 1)], N), 'live')
check('future notification incarnation', solve('live', [('connection-lost', N + 1)], N), 'live')
check('unknown event is inert', solve('live', [('unknown-event', N)], N), 'live')
check('empty delivery batch', solve('live', [], N), 'live')
check('N stale deliveries before current delivery', solve('live', [('connection-lost', N - 1)] * N + [('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('live', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('live', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-cached', [('connection-restored', N), ('snapshot-current', N)], N), 'live')
check('trace connection-restored then snapshot-failed', solve('offline-cached', [('connection-restored', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace connection-restored then connection-lost', solve('offline-cached', [('connection-restored', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-current then connection-lost', solve('reconciling', [('snapshot-current', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-failed then retry-refresh', solve('reconciling', [('snapshot-failed', N), ('retry-refresh', N)], N), 'reconciling')
check('trace retry-refresh then snapshot-current', solve('offline-error', [('retry-refresh', N), ('snapshot-current', N)], N), 'live')
check('trace retry-refresh then snapshot-failed', solve('offline-error', [('retry-refresh', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace retry-refresh then connection-lost', solve('offline-error', [('retry-refresh', N), ('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('reconciling', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('reconciling', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace open-cached-item then close-preview', solve('offline-cached', [('open-cached-item', N), ('close-preview', N)], N), 'offline-cached')
check('trace open-cached-item then connection-restored', solve('offline-cached', [('open-cached-item', N), ('connection-restored', N)], N), 'reconciling-preview')
check('trace close-preview then connection-restored', solve('offline-preview', [('close-preview', N), ('connection-restored', N)], N), 'reconciling')
check('trace close-preview then open-cached-item', solve('offline-preview', [('close-preview', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-preview', [('connection-restored', N), ('snapshot-current', N)], N), 'live-preview')
check('domain state regression 1: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 7, 'items': ['stale']})], N, {}, 'cached'), {'phase': 'reconciling', 'value': ['old']})
check('domain state regression 2: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'cached'), {'phase': 'live', 'value': ['new']})
check('domain state regression 3: cached', solve('live', [('connection-lost', N, {})], N, {'fresh': True}, 'cached'), {'phase': 'offline-cached', 'value': ['old']})
check('domain state regression 4: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
check('domain state regression 5: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['old', 'new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': 'old'})
check('domain state regression 6: fresh', solve('reconciling', [('snapshot-failed', N, {})], N, {}, 'fresh'), {'phase': 'offline-error', 'value': False})
check('domain state regression 7: preview', solve('offline-preview', [('connection-restored', N, {}), ('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
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 = {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'reconciling', ('reconciling-preview', 'snapshot-current'): 'live-preview'}
    data = json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})) if payload is None else {**json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})), **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 == 'snapshot-current':
                if arg.get('cursor',data['cursor']) < data['cursor']: event = 'unknown-event'
                else:
                    data['cursor'] = arg.get('cursor',data['cursor'])
                    data['cached'] = arg.get('items',data['cached'])[:]
            if event == 'connection-lost': data['fresh'] = False
            state = table.get((state, event), state)
            if state in ('live','live-preview'): data['fresh'] = True
            if state == 'live-preview' and data['preview'] not in data['cached']: data['preview'] = None
            if state == 'empty': data['cached'] = []
    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('live / connection-lost', solve('live', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / connection-restored', solve('offline-cached', [('connection-restored', N)], N), 'reconciling')
check('reconciling / snapshot-current', solve('reconciling', [('snapshot-current', N)], N), 'live')
check('reconciling / snapshot-failed', solve('reconciling', [('snapshot-failed', N)], N), 'offline-error')
check('offline-error / retry-refresh', solve('offline-error', [('retry-refresh', N)], N), 'reconciling')
check('reconciling / connection-lost', solve('reconciling', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / open-cached-item', solve('offline-cached', [('open-cached-item', N)], N), 'offline-preview')
check('offline-preview / close-preview', solve('offline-preview', [('close-preview', N)], N), 'offline-cached')
check('offline-preview / connection-restored', solve('offline-preview', [('connection-restored', N)], N), 'reconciling-preview')
check('reconciling-preview / snapshot-current', solve('reconciling-preview', [('snapshot-current', N)], N), 'live-preview')
check('older notification incarnation', solve('live', [('connection-lost', N - 1)], N), 'live')
check('future notification incarnation', solve('live', [('connection-lost', N + 1)], N), 'live')
check('unknown event is inert', solve('live', [('unknown-event', N)], N), 'live')
check('empty delivery batch', solve('live', [], N), 'live')
check('N stale deliveries before current delivery', solve('live', [('connection-lost', N - 1)] * N + [('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('live', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('live', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-cached', [('connection-restored', N), ('snapshot-current', N)], N), 'live')
check('trace connection-restored then snapshot-failed', solve('offline-cached', [('connection-restored', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace connection-restored then connection-lost', solve('offline-cached', [('connection-restored', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-current then connection-lost', solve('reconciling', [('snapshot-current', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-failed then retry-refresh', solve('reconciling', [('snapshot-failed', N), ('retry-refresh', N)], N), 'reconciling')
check('trace retry-refresh then snapshot-current', solve('offline-error', [('retry-refresh', N), ('snapshot-current', N)], N), 'live')
check('trace retry-refresh then snapshot-failed', solve('offline-error', [('retry-refresh', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace retry-refresh then connection-lost', solve('offline-error', [('retry-refresh', N), ('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('reconciling', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('reconciling', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace open-cached-item then close-preview', solve('offline-cached', [('open-cached-item', N), ('close-preview', N)], N), 'offline-cached')
check('trace open-cached-item then connection-restored', solve('offline-cached', [('open-cached-item', N), ('connection-restored', N)], N), 'reconciling-preview')
check('trace close-preview then connection-restored', solve('offline-preview', [('close-preview', N), ('connection-restored', N)], N), 'reconciling')
check('trace close-preview then open-cached-item', solve('offline-preview', [('close-preview', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-preview', [('connection-restored', N), ('snapshot-current', N)], N), 'live-preview')
check('domain state regression 1: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 7, 'items': ['stale']})], N, {}, 'cached'), {'phase': 'reconciling', 'value': ['old']})
check('domain state regression 2: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'cached'), {'phase': 'live', 'value': ['new']})
check('domain state regression 3: cached', solve('live', [('connection-lost', N, {})], N, {'fresh': True}, 'cached'), {'phase': 'offline-cached', 'value': ['old']})
check('domain state regression 4: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
check('domain state regression 5: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['old', 'new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': 'old'})
check('domain state regression 6: fresh', solve('reconciling', [('snapshot-failed', N, {})], N, {}, 'fresh'), {'phase': 'offline-error', 'value': False})
check('domain state regression 7: preview', solve('offline-preview', [('connection-restored', N, {}), ('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
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 connection-restored event leaves a notification in live instead of reconciling-preview.

Notification inbox reconnect recovery is a bounded visual-notification workflow with mutable policy state {'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False}. Its default transition relation is {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'reconciling-preview', ('reconciling-preview', 'snapshot-current'): 'live-preview'}; 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 = {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'reconciling-preview', ('reconciling-preview', 'snapshot-current'): 'live'}
    data = json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})) if payload is None else {**json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})), **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 == 'snapshot-current':
                if arg.get('cursor',data['cursor']) < data['cursor']: event = 'unknown-event'
                else:
                    data['cursor'] = arg.get('cursor',data['cursor'])
                    data['cached'] = arg.get('items',data['cached'])[:]
            if event == 'connection-lost': data['fresh'] = False
            state = table.get((state, event), state)
            if state in ('live','live-preview'): data['fresh'] = True
            if state == 'live-preview' and data['preview'] not in data['cached']: data['preview'] = None
            if state == 'empty': data['cached'] = []
    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('live / connection-lost', solve('live', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / connection-restored', solve('offline-cached', [('connection-restored', N)], N), 'reconciling')
check('reconciling / snapshot-current', solve('reconciling', [('snapshot-current', N)], N), 'live')
check('reconciling / snapshot-failed', solve('reconciling', [('snapshot-failed', N)], N), 'offline-error')
check('offline-error / retry-refresh', solve('offline-error', [('retry-refresh', N)], N), 'reconciling')
check('reconciling / connection-lost', solve('reconciling', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / open-cached-item', solve('offline-cached', [('open-cached-item', N)], N), 'offline-preview')
check('offline-preview / close-preview', solve('offline-preview', [('close-preview', N)], N), 'offline-cached')
check('offline-preview / connection-restored', solve('offline-preview', [('connection-restored', N)], N), 'reconciling-preview')
check('reconciling-preview / snapshot-current', solve('reconciling-preview', [('snapshot-current', N)], N), 'live-preview')
check('older notification incarnation', solve('live', [('connection-lost', N - 1)], N), 'live')
check('future notification incarnation', solve('live', [('connection-lost', N + 1)], N), 'live')
check('unknown event is inert', solve('live', [('unknown-event', N)], N), 'live')
check('empty delivery batch', solve('live', [], N), 'live')
check('N stale deliveries before current delivery', solve('live', [('connection-lost', N - 1)] * N + [('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('live', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('live', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-cached', [('connection-restored', N), ('snapshot-current', N)], N), 'live')
check('trace connection-restored then snapshot-failed', solve('offline-cached', [('connection-restored', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace connection-restored then connection-lost', solve('offline-cached', [('connection-restored', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-current then connection-lost', solve('reconciling', [('snapshot-current', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-failed then retry-refresh', solve('reconciling', [('snapshot-failed', N), ('retry-refresh', N)], N), 'reconciling')
check('trace retry-refresh then snapshot-current', solve('offline-error', [('retry-refresh', N), ('snapshot-current', N)], N), 'live')
check('trace retry-refresh then snapshot-failed', solve('offline-error', [('retry-refresh', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace retry-refresh then connection-lost', solve('offline-error', [('retry-refresh', N), ('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('reconciling', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('reconciling', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace open-cached-item then close-preview', solve('offline-cached', [('open-cached-item', N), ('close-preview', N)], N), 'offline-cached')
check('trace open-cached-item then connection-restored', solve('offline-cached', [('open-cached-item', N), ('connection-restored', N)], N), 'reconciling-preview')
check('trace close-preview then connection-restored', solve('offline-preview', [('close-preview', N), ('connection-restored', N)], N), 'reconciling')
check('trace close-preview then open-cached-item', solve('offline-preview', [('close-preview', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-preview', [('connection-restored', N), ('snapshot-current', N)], N), 'live-preview')
check('domain state regression 1: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 7, 'items': ['stale']})], N, {}, 'cached'), {'phase': 'reconciling', 'value': ['old']})
check('domain state regression 2: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'cached'), {'phase': 'live', 'value': ['new']})
check('domain state regression 3: cached', solve('live', [('connection-lost', N, {})], N, {'fresh': True}, 'cached'), {'phase': 'offline-cached', 'value': ['old']})
check('domain state regression 4: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
check('domain state regression 5: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['old', 'new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': 'old'})
check('domain state regression 6: fresh', solve('reconciling', [('snapshot-failed', N, {})], N, {}, 'fresh'), {'phase': 'offline-error', 'value': False})
check('domain state regression 7: preview', solve('offline-preview', [('connection-restored', N, {}), ('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
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 = {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'reconciling-preview', ('reconciling-preview', 'snapshot-current'): 'offline-preview'}
    data = json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})) if payload is None else {**json.loads(json.dumps({'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False})), **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 == 'snapshot-current':
                if arg.get('cursor',data['cursor']) < data['cursor']: event = 'unknown-event'
                else:
                    data['cursor'] = arg.get('cursor',data['cursor'])
                    data['cached'] = arg.get('items',data['cached'])[:]
            if event == 'connection-lost': data['fresh'] = False
            state = table.get((state, event), state)
            if state in ('live','live-preview'): data['fresh'] = True
            if state == 'live-preview' and data['preview'] not in data['cached']: data['preview'] = None
            if state == 'empty': data['cached'] = []
    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('live / connection-lost', solve('live', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / connection-restored', solve('offline-cached', [('connection-restored', N)], N), 'reconciling')
check('reconciling / snapshot-current', solve('reconciling', [('snapshot-current', N)], N), 'live')
check('reconciling / snapshot-failed', solve('reconciling', [('snapshot-failed', N)], N), 'offline-error')
check('offline-error / retry-refresh', solve('offline-error', [('retry-refresh', N)], N), 'reconciling')
check('reconciling / connection-lost', solve('reconciling', [('connection-lost', N)], N), 'offline-cached')
check('offline-cached / open-cached-item', solve('offline-cached', [('open-cached-item', N)], N), 'offline-preview')
check('offline-preview / close-preview', solve('offline-preview', [('close-preview', N)], N), 'offline-cached')
check('offline-preview / connection-restored', solve('offline-preview', [('connection-restored', N)], N), 'reconciling-preview')
check('reconciling-preview / snapshot-current', solve('reconciling-preview', [('snapshot-current', N)], N), 'live-preview')
check('older notification incarnation', solve('live', [('connection-lost', N - 1)], N), 'live')
check('future notification incarnation', solve('live', [('connection-lost', N + 1)], N), 'live')
check('unknown event is inert', solve('live', [('unknown-event', N)], N), 'live')
check('empty delivery batch', solve('live', [], N), 'live')
check('N stale deliveries before current delivery', solve('live', [('connection-lost', N - 1)] * N + [('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('live', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('live', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-cached', [('connection-restored', N), ('snapshot-current', N)], N), 'live')
check('trace connection-restored then snapshot-failed', solve('offline-cached', [('connection-restored', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace connection-restored then connection-lost', solve('offline-cached', [('connection-restored', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-current then connection-lost', solve('reconciling', [('snapshot-current', N), ('connection-lost', N)], N), 'offline-cached')
check('trace snapshot-failed then retry-refresh', solve('reconciling', [('snapshot-failed', N), ('retry-refresh', N)], N), 'reconciling')
check('trace retry-refresh then snapshot-current', solve('offline-error', [('retry-refresh', N), ('snapshot-current', N)], N), 'live')
check('trace retry-refresh then snapshot-failed', solve('offline-error', [('retry-refresh', N), ('snapshot-failed', N)], N), 'offline-error')
check('trace retry-refresh then connection-lost', solve('offline-error', [('retry-refresh', N), ('connection-lost', N)], N), 'offline-cached')
check('trace connection-lost then connection-restored', solve('reconciling', [('connection-lost', N), ('connection-restored', N)], N), 'reconciling')
check('trace connection-lost then open-cached-item', solve('reconciling', [('connection-lost', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace open-cached-item then close-preview', solve('offline-cached', [('open-cached-item', N), ('close-preview', N)], N), 'offline-cached')
check('trace open-cached-item then connection-restored', solve('offline-cached', [('open-cached-item', N), ('connection-restored', N)], N), 'reconciling-preview')
check('trace close-preview then connection-restored', solve('offline-preview', [('close-preview', N), ('connection-restored', N)], N), 'reconciling')
check('trace close-preview then open-cached-item', solve('offline-preview', [('close-preview', N), ('open-cached-item', N)], N), 'offline-preview')
check('trace connection-restored then snapshot-current', solve('offline-preview', [('connection-restored', N), ('snapshot-current', N)], N), 'live-preview')
check('domain state regression 1: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 7, 'items': ['stale']})], N, {}, 'cached'), {'phase': 'reconciling', 'value': ['old']})
check('domain state regression 2: cached', solve('reconciling', [('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'cached'), {'phase': 'live', 'value': ['new']})
check('domain state regression 3: cached', solve('live', [('connection-lost', N, {})], N, {'fresh': True}, 'cached'), {'phase': 'offline-cached', 'value': ['old']})
check('domain state regression 4: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
check('domain state regression 5: preview', solve('reconciling-preview', [('snapshot-current', N, {'items': ['old', 'new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': 'old'})
check('domain state regression 6: fresh', solve('reconciling', [('snapshot-failed', N, {})], N, {}, 'fresh'), {'phase': 'offline-error', 'value': False})
check('domain state regression 7: preview', solve('offline-preview', [('connection-restored', N, {}), ('snapshot-current', N, {'cursor': 9, 'items': ['new']})], N, {}, 'preview'), {'phase': 'live-preview', 'value': None})
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 snapshot-current event leaves a notification in live instead of live-preview.

Notification inbox reconnect recovery is a bounded visual-notification workflow with mutable policy state {'cached': ['old'], 'cursor': 8, 'preview': 'old', 'fresh': False}. Its default transition relation is {('live', 'connection-lost'): 'offline-cached', ('offline-cached', 'connection-restored'): 'reconciling', ('reconciling', 'snapshot-current'): 'live', ('reconciling', 'snapshot-failed'): 'offline-error', ('offline-error', 'retry-refresh'): 'reconciling', ('reconciling', 'connection-lost'): 'offline-cached', ('offline-cached', 'open-cached-item'): 'offline-preview', ('offline-preview', 'close-preview'): 'offline-cached', ('offline-preview', 'connection-restored'): 'reconciling-preview', ('reconciling-preview', 'snapshot-current'): 'live-preview'}; 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 = {('standby', 'request-display'): 'displaying', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 = {('standby', 'request-display'): 'standby', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 request-display event leaves a notification in displaying instead of claiming.

Single-surface notification display ownership is a bounded visual-notification workflow with mutable policy state {'token': 5, 'owner': 'local', 'mounted': False, 'released': []}. Its default transition relation is {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}; 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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'standby', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'claiming', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 lease-granted event leaves a notification in standby instead of displaying.

Single-surface notification display ownership is a bounded visual-notification workflow with mutable policy state {'token': 5, 'owner': 'local', 'mounted': False, 'released': []}. Its default transition relation is {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}; 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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'displaying', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'releasing', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 lease-denied event leaves a notification in displaying instead of standby.

Single-surface notification display ownership is a bounded visual-notification workflow with mutable policy state {'token': 5, 'owner': 'local', 'mounted': False, 'released': []}. Its default transition relation is {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}; 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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'displaying', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 = {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'standby', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}
    data = json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})) if payload is None else {**json.loads(json.dumps({'token': 5, 'owner': 'local', 'mounted': False, 'released': []})), **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 ('lease-granted','lease-revoked','release-confirmed','late-lease-granted') and arg.get('token',data['token']) != data['token']: event = 'unknown-event'
            if event == 'foreign-release' and arg.get('owner','foreign') == data['owner']: event = 'release-confirmed'
            state = table.get((state, event), state)
            if state == 'displaying': data['mounted'] = True
            if state in ('standby','withdrawing'): data['mounted'] = False
            if state == 'releasing' and data['token'] not in data['released']: data['released'].append(data['token'])
    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('standby / request-display', solve('standby', [('request-display', N)], N), 'claiming')
check('claiming / lease-granted', solve('claiming', [('lease-granted', N)], N), 'displaying')
check('claiming / lease-denied', solve('claiming', [('lease-denied', N)], N), 'standby')
check('displaying / lease-revoked', solve('displaying', [('lease-revoked', N)], N), 'withdrawing')
check('withdrawing / unmounted', solve('withdrawing', [('unmounted', N)], N), 'standby')
check('displaying / close-surface', solve('displaying', [('close-surface', N)], N), 'releasing')
check('releasing / release-confirmed', solve('releasing', [('release-confirmed', N)], N), 'standby')
check('claiming / cancel-display', solve('claiming', [('cancel-display', N)], N), 'claim-canceled')
check('claim-canceled / late-lease-granted', solve('claim-canceled', [('late-lease-granted', N)], N), 'releasing')
check('standby / foreign-release', solve('standby', [('foreign-release', N)], N), 'standby')
check('older notification incarnation', solve('standby', [('request-display', N - 1)], N), 'standby')
check('future notification incarnation', solve('standby', [('request-display', N + 1)], N), 'standby')
check('unknown event is inert', solve('standby', [('unknown-event', N)], N), 'standby')
check('empty delivery batch', solve('standby', [], N), 'standby')
check('N stale deliveries before current delivery', solve('standby', [('request-display', N - 1)] * N + [('request-display', N)], N), 'claiming')
check('trace request-display then lease-granted', solve('standby', [('request-display', N), ('lease-granted', N)], N), 'displaying')
check('trace request-display then lease-denied', solve('standby', [('request-display', N), ('lease-denied', N)], N), 'standby')
check('trace request-display then cancel-display', solve('standby', [('request-display', N), ('cancel-display', N)], N), 'claim-canceled')
check('trace lease-granted then lease-revoked', solve('claiming', [('lease-granted', N), ('lease-revoked', N)], N), 'withdrawing')
check('trace lease-granted then close-surface', solve('claiming', [('lease-granted', N), ('close-surface', N)], N), 'releasing')
check('trace lease-denied then request-display', solve('claiming', [('lease-denied', N), ('request-display', N)], N), 'claiming')
check('trace lease-denied then foreign-release', solve('claiming', [('lease-denied', N), ('foreign-release', N)], N), 'standby')
check('trace lease-revoked then unmounted', solve('displaying', [('lease-revoked', N), ('unmounted', N)], N), 'standby')
check('trace unmounted then request-display', solve('withdrawing', [('unmounted', N), ('request-display', N)], N), 'claiming')
check('trace unmounted then foreign-release', solve('withdrawing', [('unmounted', N), ('foreign-release', N)], N), 'standby')
check('trace close-surface then release-confirmed', solve('displaying', [('close-surface', N), ('release-confirmed', N)], N), 'standby')
check('trace release-confirmed then request-display', solve('releasing', [('release-confirmed', N), ('request-display', N)], N), 'claiming')
check('trace release-confirmed then foreign-release', solve('releasing', [('release-confirmed', N), ('foreign-release', N)], N), 'standby')
check('trace cancel-display then late-lease-granted', solve('claiming', [('cancel-display', N), ('late-lease-granted', N)], N), 'releasing')
check('trace late-lease-granted then release-confirmed', solve('claim-canceled', [('late-lease-granted', N), ('release-confirmed', N)], N), 'standby')
check('trace foreign-release then request-display', solve('standby', [('foreign-release', N), ('request-display', N)], N), 'claiming')
check('trace foreign-release then foreign-release', solve('standby', [('foreign-release', N), ('foreign-release', N)], N), 'standby')
check('domain state regression 1: mounted', solve('claiming', [('lease-granted', N, {'token': 4})], N, {}, 'mounted'), {'phase': 'claiming', 'value': False})
check('domain state regression 2: mounted', solve('claiming', [('lease-granted', N, {'token': 5})], N, {}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 3: mounted', solve('displaying', [('lease-revoked', N, {'token': 4})], N, {'mounted': True}, 'mounted'), {'phase': 'displaying', 'value': True})
check('domain state regression 4: mounted', solve('displaying', [('lease-revoked', N, {'token': 5})], N, {'mounted': True}, 'mounted'), {'phase': 'withdrawing', 'value': False})
check('domain state regression 5: released', solve('claim-canceled', [('late-lease-granted', N, {})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
check('domain state regression 6: released', solve('standby', [('foreign-release', N, {'owner': 'foreign'})], N, {}, 'released'), {'phase': 'standby', 'value': []})
check('domain state regression 7: released', solve('standby', [('request-display', N, {}), ('cancel-display', N, {}), ('late-lease-granted', N, {'token': 5})], N, {}, 'released'), {'phase': 'releasing', 'value': [5]})
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 lease-revoked event leaves a notification in displaying instead of withdrawing.

Single-surface notification display ownership is a bounded visual-notification workflow with mutable policy state {'token': 5, 'owner': 'local', 'mounted': False, 'released': []}. Its default transition relation is {('standby', 'request-display'): 'claiming', ('claiming', 'lease-granted'): 'displaying', ('claiming', 'lease-denied'): 'standby', ('displaying', 'lease-revoked'): 'withdrawing', ('withdrawing', 'unmounted'): 'standby', ('displaying', 'close-surface'): 'releasing', ('releasing', 'release-confirmed'): 'standby', ('claiming', 'cancel-display'): 'claim-canceled', ('claim-canceled', 'late-lease-granted'): 'releasing', ('standby', 'foreign-release'): 'standby'}; 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