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 = {('account-a-visible', 'switch-to-b'): 'scrubbing-a', ('scrubbing-a', 'scrub-complete'): 'account-b-loading', ('account-b-loading', 'b-inbox-loaded'): 'account-b-visible', ('account-b-loading', 'b-inbox-failed'): 'switch-error', ('account-b-loading', 'late-a-response'): 'account-b-loading', ('switch-error', 'retry-b-load'): 'account-b-loading', ('account-b-visible', 'sign-out'): 'scrubbing-b', ('scrubbing-b', 'scrub-complete'): 'signed-out', ('signed-out', 'late-b-response'): 'signed-out', ('scrubbing-a', 'switch-back-to-a'): 'account-a-visible'}
    data = json.loads(json.dumps({'owner': 'a', 'rows': ['a-notice'], 'target': 'b', 'scrubbed': False})) if payload is None else {**json.loads(json.dumps({'owner': 'a', 'rows': ['a-notice'], 'target': 'b', 'scrubbed': 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 == 'b-inbox-loaded' and arg.get('owner',data['target']) != data['target']: event = 'unknown-event'
            if event == 'b-inbox-loaded' and not data['scrubbed'] and arg.get('require_scrub',False): event = 'unknown-event'
            if event == 'switch-back-to-a': data['target'] = 'a'
            state = table.get((state, event), state)
            if state.startswith('scrubbing'):
                data['rows'] = []
                data['scrubbed'] = False
            if event == 'scrub-complete': data['scrubbed'] = True
            if state == 'account-b-visible':
                data['owner'] = 'b'
                data['rows'] = arg.get('rows',['b-notice'])[:]
            if state == 'signed-out':
                data['owner'] = None
                data['rows'] = []
    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('account-a-visible / switch-to-b', solve('account-a-visible', [('switch-to-b', N)], N), 'scrubbing-a')
check('scrubbing-a / scrub-complete', solve('scrubbing-a', [('scrub-complete', N)], N), 'account-b-loading')
check('account-b-loading / b-inbox-loaded', solve('account-b-loading', [('b-inbox-loaded', N)], N), 'account-b-visible')
check('account-b-loading / b-inbox-failed', solve('account-b-loading', [('b-inbox-failed', N)], N), 'switch-error')
check('account-b-loading / late-a-response', solve('account-b-loading', [('late-a-response', N)], N), 'account-b-loading')
check('switch-error / retry-b-load', solve('switch-error', [('retry-b-load', N)], N), 'account-b-loading')
check('account-b-visible / sign-out', solve('account-b-visible', [('sign-out', N)], N), 'scrubbing-b')
check('scrubbing-b / scrub-complete', solve('scrubbing-b', [('scrub-complete', N)], N), 'signed-out')
check('signed-out / late-b-response', solve('signed-out', [('late-b-response', N)], N), 'signed-out')
check('scrubbing-a / switch-back-to-a', solve('scrubbing-a', [('switch-back-to-a', N)], N), 'scrubbing-for-a')
check('older notification incarnation', solve('account-a-visible', [('switch-to-b', N - 1)], N), 'account-a-visible')
check('future notification incarnation', solve('account-a-visible', [('switch-to-b', N + 1)], N), 'account-a-visible')
check('unknown event is inert', solve('account-a-visible', [('unknown-event', N)], N), 'account-a-visible')
check('empty delivery batch', solve('account-a-visible', [], N), 'account-a-visible')
check('N stale deliveries before current delivery', solve('account-a-visible', [('switch-to-b', N - 1)] * N + [('switch-to-b', N)], N), 'scrubbing-a')
check('trace switch-to-b then scrub-complete', solve('account-a-visible', [('switch-to-b', N), ('scrub-complete', N)], N), 'account-b-loading')
check('trace switch-to-b then switch-back-to-a', solve('account-a-visible', [('switch-to-b', N), ('switch-back-to-a', N)], N), 'scrubbing-for-a')
check('trace scrub-complete then b-inbox-loaded', solve('scrubbing-a', [('scrub-complete', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace scrub-complete then b-inbox-failed', solve('scrubbing-a', [('scrub-complete', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace scrub-complete then late-a-response', solve('scrubbing-a', [('scrub-complete', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace b-inbox-loaded then sign-out', solve('account-b-loading', [('b-inbox-loaded', N), ('sign-out', N)], N), 'scrubbing-b')
check('trace b-inbox-failed then retry-b-load', solve('account-b-loading', [('b-inbox-failed', N), ('retry-b-load', N)], N), 'account-b-loading')
check('trace late-a-response then b-inbox-loaded', solve('account-b-loading', [('late-a-response', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace late-a-response then b-inbox-failed', solve('account-b-loading', [('late-a-response', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace late-a-response then late-a-response', solve('account-b-loading', [('late-a-response', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace retry-b-load then b-inbox-loaded', solve('switch-error', [('retry-b-load', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace retry-b-load then b-inbox-failed', solve('switch-error', [('retry-b-load', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace retry-b-load then late-a-response', solve('switch-error', [('retry-b-load', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace sign-out then scrub-complete', solve('account-b-visible', [('sign-out', N), ('scrub-complete', N)], N), 'signed-out')
check('trace scrub-complete then late-b-response', solve('scrubbing-b', [('scrub-complete', N), ('late-b-response', N)], N), 'signed-out')
check('trace late-b-response then late-b-response', solve('signed-out', [('late-b-response', N), ('late-b-response', N)], N), 'signed-out')
check('domain state regression 1: rows', solve('account-a-visible', [('switch-to-b', N, {})], N, {}, 'rows'), {'phase': 'scrubbing-a', 'value': []})
check('domain state regression 2: rows', solve('account-b-loading', [('b-inbox-loaded', N, {'owner': 'a'})], N, {}, 'rows'), {'phase': 'account-b-loading', 'value': ['a-notice']})
check('domain state regression 3: owner', solve('account-b-loading', [('b-inbox-loaded', N, {'require_scrub': True})], N, {}, 'owner'), {'phase': 'account-b-loading', 'value': 'a'})
check('domain state regression 4: rows', solve('account-b-loading', [('b-inbox-loaded', N, {'require_scrub': True})], N, {'scrubbed': True, 'rows': []}, 'rows'), {'phase': 'account-b-visible', 'value': ['b-notice']})
check('domain state regression 5: owner', solve('scrubbing-b', [('scrub-complete', N, {})], N, {'owner': 'b'}, 'owner'), {'phase': 'signed-out', 'value': None})
check('domain state regression 6: target', solve('scrubbing-a', [('switch-back-to-a', N, {})], N, {}, 'target'), {'phase': 'scrubbing-for-a', 'value': 'a'})
check('domain state regression 7: rows', solve('account-a-visible', [('switch-to-b', N, {}), ('scrub-complete', N, {}), ('b-inbox-loaded', N, {'require_scrub': True, 'rows': ['b']}), ('sign-out', N, {}), ('scrub-complete', N, {}), ('late-b-response', N, {})], N, {}, 'rows'), {'phase': 'signed-out', 'value': []})
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 = {('account-a-visible', 'switch-to-b'): 'scrubbing-a', ('scrubbing-a', 'scrub-complete'): 'account-b-loading', ('account-b-loading', 'b-inbox-loaded'): 'account-b-visible', ('account-b-loading', 'b-inbox-failed'): 'switch-error', ('account-b-loading', 'late-a-response'): 'account-b-loading', ('switch-error', 'retry-b-load'): 'account-b-loading', ('account-b-visible', 'sign-out'): 'scrubbing-b', ('scrubbing-b', 'scrub-complete'): 'signed-out', ('signed-out', 'late-b-response'): 'signed-out', ('scrubbing-a', 'switch-back-to-a'): 'account-b-loading'}
    data = json.loads(json.dumps({'owner': 'a', 'rows': ['a-notice'], 'target': 'b', 'scrubbed': False})) if payload is None else {**json.loads(json.dumps({'owner': 'a', 'rows': ['a-notice'], 'target': 'b', 'scrubbed': 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 == 'b-inbox-loaded' and arg.get('owner',data['target']) != data['target']: event = 'unknown-event'
            if event == 'b-inbox-loaded' and not data['scrubbed'] and arg.get('require_scrub',False): event = 'unknown-event'
            if event == 'switch-back-to-a': data['target'] = 'a'
            state = table.get((state, event), state)
            if state.startswith('scrubbing'):
                data['rows'] = []
                data['scrubbed'] = False
            if event == 'scrub-complete': data['scrubbed'] = True
            if state == 'account-b-visible':
                data['owner'] = 'b'
                data['rows'] = arg.get('rows',['b-notice'])[:]
            if state == 'signed-out':
                data['owner'] = None
                data['rows'] = []
    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('account-a-visible / switch-to-b', solve('account-a-visible', [('switch-to-b', N)], N), 'scrubbing-a')
check('scrubbing-a / scrub-complete', solve('scrubbing-a', [('scrub-complete', N)], N), 'account-b-loading')
check('account-b-loading / b-inbox-loaded', solve('account-b-loading', [('b-inbox-loaded', N)], N), 'account-b-visible')
check('account-b-loading / b-inbox-failed', solve('account-b-loading', [('b-inbox-failed', N)], N), 'switch-error')
check('account-b-loading / late-a-response', solve('account-b-loading', [('late-a-response', N)], N), 'account-b-loading')
check('switch-error / retry-b-load', solve('switch-error', [('retry-b-load', N)], N), 'account-b-loading')
check('account-b-visible / sign-out', solve('account-b-visible', [('sign-out', N)], N), 'scrubbing-b')
check('scrubbing-b / scrub-complete', solve('scrubbing-b', [('scrub-complete', N)], N), 'signed-out')
check('signed-out / late-b-response', solve('signed-out', [('late-b-response', N)], N), 'signed-out')
check('scrubbing-a / switch-back-to-a', solve('scrubbing-a', [('switch-back-to-a', N)], N), 'scrubbing-for-a')
check('older notification incarnation', solve('account-a-visible', [('switch-to-b', N - 1)], N), 'account-a-visible')
check('future notification incarnation', solve('account-a-visible', [('switch-to-b', N + 1)], N), 'account-a-visible')
check('unknown event is inert', solve('account-a-visible', [('unknown-event', N)], N), 'account-a-visible')
check('empty delivery batch', solve('account-a-visible', [], N), 'account-a-visible')
check('N stale deliveries before current delivery', solve('account-a-visible', [('switch-to-b', N - 1)] * N + [('switch-to-b', N)], N), 'scrubbing-a')
check('trace switch-to-b then scrub-complete', solve('account-a-visible', [('switch-to-b', N), ('scrub-complete', N)], N), 'account-b-loading')
check('trace switch-to-b then switch-back-to-a', solve('account-a-visible', [('switch-to-b', N), ('switch-back-to-a', N)], N), 'scrubbing-for-a')
check('trace scrub-complete then b-inbox-loaded', solve('scrubbing-a', [('scrub-complete', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace scrub-complete then b-inbox-failed', solve('scrubbing-a', [('scrub-complete', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace scrub-complete then late-a-response', solve('scrubbing-a', [('scrub-complete', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace b-inbox-loaded then sign-out', solve('account-b-loading', [('b-inbox-loaded', N), ('sign-out', N)], N), 'scrubbing-b')
check('trace b-inbox-failed then retry-b-load', solve('account-b-loading', [('b-inbox-failed', N), ('retry-b-load', N)], N), 'account-b-loading')
check('trace late-a-response then b-inbox-loaded', solve('account-b-loading', [('late-a-response', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace late-a-response then b-inbox-failed', solve('account-b-loading', [('late-a-response', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace late-a-response then late-a-response', solve('account-b-loading', [('late-a-response', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace retry-b-load then b-inbox-loaded', solve('switch-error', [('retry-b-load', N), ('b-inbox-loaded', N)], N), 'account-b-visible')
check('trace retry-b-load then b-inbox-failed', solve('switch-error', [('retry-b-load', N), ('b-inbox-failed', N)], N), 'switch-error')
check('trace retry-b-load then late-a-response', solve('switch-error', [('retry-b-load', N), ('late-a-response', N)], N), 'account-b-loading')
check('trace sign-out then scrub-complete', solve('account-b-visible', [('sign-out', N), ('scrub-complete', N)], N), 'signed-out')
check('trace scrub-complete then late-b-response', solve('scrubbing-b', [('scrub-complete', N), ('late-b-response', N)], N), 'signed-out')
check('trace late-b-response then late-b-response', solve('signed-out', [('late-b-response', N), ('late-b-response', N)], N), 'signed-out')
check('domain state regression 1: rows', solve('account-a-visible', [('switch-to-b', N, {})], N, {}, 'rows'), {'phase': 'scrubbing-a', 'value': []})
check('domain state regression 2: rows', solve('account-b-loading', [('b-inbox-loaded', N, {'owner': 'a'})], N, {}, 'rows'), {'phase': 'account-b-loading', 'value': ['a-notice']})
check('domain state regression 3: owner', solve('account-b-loading', [('b-inbox-loaded', N, {'require_scrub': True})], N, {}, 'owner'), {'phase': 'account-b-loading', 'value': 'a'})
check('domain state regression 4: rows', solve('account-b-loading', [('b-inbox-loaded', N, {'require_scrub': True})], N, {'scrubbed': True, 'rows': []}, 'rows'), {'phase': 'account-b-visible', 'value': ['b-notice']})
check('domain state regression 5: owner', solve('scrubbing-b', [('scrub-complete', N, {})], N, {'owner': 'b'}, 'owner'), {'phase': 'signed-out', 'value': None})
check('domain state regression 6: target', solve('scrubbing-a', [('switch-back-to-a', N, {})], N, {}, 'target'), {'phase': 'scrubbing-for-a', 'value': 'a'})
check('domain state regression 7: rows', solve('account-a-visible', [('switch-to-b', N, {}), ('scrub-complete', N, {}), ('b-inbox-loaded', N, {'require_scrub': True, 'rows': ['b']}), ('sign-out', N, {}), ('scrub-complete', N, {}), ('late-b-response', N, {})], N, {}, 'rows'), {'phase': 'signed-out', 'value': []})
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 switch-back-to-a event leaves a notification in account-a-visible instead of scrubbing-for-a.

Notification surface account switch is a bounded visual-notification workflow with mutable policy state {'owner': 'a', 'rows': ['a-notice'], 'target': 'b', 'scrubbed': False}. Its default transition relation is {('account-a-visible', 'switch-to-b'): 'scrubbing-a', ('scrubbing-a', 'scrub-complete'): 'account-b-loading', ('account-b-loading', 'b-inbox-loaded'): 'account-b-visible', ('account-b-loading', 'b-inbox-failed'): 'switch-error', ('account-b-loading', 'late-a-response'): 'account-b-loading', ('switch-error', 'retry-b-load'): 'account-b-loading', ('account-b-visible', 'sign-out'): 'scrubbing-b', ('scrubbing-b', 'scrub-complete'): 'signed-out', ('signed-out', 'late-b-response'): 'signed-out', ('scrubbing-a', 'switch-back-to-a'): 'scrubbing-for-a'}; 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 = {('unpinned', 'pin-click'): 'pinned', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 = {('unpinned', 'pin-click'): 'unpinned', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 pin-click event leaves a notification in pinned instead of pin-pending.

Inbox notification pin persistence is a bounded visual-notification workflow with mutable policy state {'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True}. Its default transition relation is {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}; 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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'unpinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pin-error', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 pin-success event leaves a notification in unpinned instead of pinned.

Inbox notification pin persistence is a bounded visual-notification workflow with mutable policy state {'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True}. Its default transition relation is {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}; 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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pinned', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'unpinned', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 pin-failure event leaves a notification in pinned instead of pin-error.

Inbox notification pin persistence is a bounded visual-notification workflow with mutable policy state {'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True}. Its default transition relation is {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}; 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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpinned', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'pinned', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 unpin-click event leaves a notification in unpinned instead of unpin-pending.

Inbox notification pin persistence is a bounded visual-notification workflow with mutable policy state {'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True}. Its default transition relation is {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}; 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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'pinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 = {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpin-error', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}
    data = json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})) if payload is None else {**json.loads(json.dumps({'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'pin-click' and data['item'] not in data['pinned'] and len(data['pinned']) >= data['capacity']: event = 'unknown-event'
            if event == 'notification-expires': data['expired'] = True
            state = table.get((state, event), state)
            if state in ('pinned','expired-pinned') and data['item'] not in data['pinned']: data['pinned'].append(data['item'])
            if state in ('unpinned','gone'): data['pinned'] = [x for x in data['pinned'] if x != data['item']]
            if state == 'gone': data['retained'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('unpinned / pin-click', solve('unpinned', [('pin-click', N)], N), 'pin-pending')
check('pin-pending / pin-success', solve('pin-pending', [('pin-success', N)], N), 'pinned')
check('pin-pending / pin-failure', solve('pin-pending', [('pin-failure', N)], N), 'pin-error')
check('pinned / unpin-click', solve('pinned', [('unpin-click', N)], N), 'unpin-pending')
check('unpin-pending / unpin-success', solve('unpin-pending', [('unpin-success', N)], N), 'unpinned')
check('unpin-pending / unpin-failure', solve('unpin-pending', [('unpin-failure', N)], N), 'unpin-error')
check('pin-error / cancel-pin-intent', solve('pin-error', [('cancel-pin-intent', N)], N), 'unpinned')
check('unpin-error / cancel-unpin-intent', solve('unpin-error', [('cancel-unpin-intent', N)], N), 'pinned')
check('pinned / notification-expires', solve('pinned', [('notification-expires', N)], N), 'expired-pinned')
check('expired-pinned / unpin-expired', solve('expired-pinned', [('unpin-expired', N)], N), 'gone')
check('older notification incarnation', solve('unpinned', [('pin-click', N - 1)], N), 'unpinned')
check('future notification incarnation', solve('unpinned', [('pin-click', N + 1)], N), 'unpinned')
check('unknown event is inert', solve('unpinned', [('unknown-event', N)], N), 'unpinned')
check('empty delivery batch', solve('unpinned', [], N), 'unpinned')
check('N stale deliveries before current delivery', solve('unpinned', [('pin-click', N - 1)] * N + [('pin-click', N)], N), 'pin-pending')
check('trace pin-click then pin-success', solve('unpinned', [('pin-click', N), ('pin-success', N)], N), 'pinned')
check('trace pin-click then pin-failure', solve('unpinned', [('pin-click', N), ('pin-failure', N)], N), 'pin-error')
check('trace pin-success then unpin-click', solve('pin-pending', [('pin-success', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace pin-success then notification-expires', solve('pin-pending', [('pin-success', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace pin-failure then cancel-pin-intent', solve('pin-pending', [('pin-failure', N), ('cancel-pin-intent', N)], N), 'unpinned')
check('trace unpin-click then unpin-success', solve('pinned', [('unpin-click', N), ('unpin-success', N)], N), 'unpinned')
check('trace unpin-click then unpin-failure', solve('pinned', [('unpin-click', N), ('unpin-failure', N)], N), 'unpin-error')
check('trace unpin-success then pin-click', solve('unpin-pending', [('unpin-success', N), ('pin-click', N)], N), 'pin-pending')
check('trace unpin-failure then cancel-unpin-intent', solve('unpin-pending', [('unpin-failure', N), ('cancel-unpin-intent', N)], N), 'pinned')
check('trace cancel-pin-intent then pin-click', solve('pin-error', [('cancel-pin-intent', N), ('pin-click', N)], N), 'pin-pending')
check('trace cancel-unpin-intent then unpin-click', solve('unpin-error', [('cancel-unpin-intent', N), ('unpin-click', N)], N), 'unpin-pending')
check('trace cancel-unpin-intent then notification-expires', solve('unpin-error', [('cancel-unpin-intent', N), ('notification-expires', N)], N), 'expired-pinned')
check('trace notification-expires then unpin-expired', solve('pinned', [('notification-expires', N), ('unpin-expired', N)], N), 'gone')
check('domain state regression 1: pinned', solve('unpinned', [('pin-click', N, {})], N, {'capacity': 1}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 2: pinned', solve('pin-pending', [('pin-success', N, {})], N, {}, 'pinned'), {'phase': 'pinned', 'value': ['other', 'n']})
check('domain state regression 3: pinned', solve('unpin-pending', [('unpin-success', N, {})], N, {'pinned': ['n', 'other']}, 'pinned'), {'phase': 'unpinned', 'value': ['other']})
check('domain state regression 4: retained', solve('pinned', [('notification-expires', N, {})], N, {'pinned': ['n']}, 'retained'), {'phase': 'expired-pinned', 'value': True})
check('domain state regression 5: pinned', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n', 'other'], 'expired': True}, 'pinned'), {'phase': 'gone', 'value': ['other']})
check('domain state regression 6: retained', solve('expired-pinned', [('unpin-expired', N, {})], N, {'pinned': ['n'], 'expired': True}, 'retained'), {'phase': 'gone', 'value': False})
check('domain state regression 7: pinned', solve('unpinned', [('pin-click', N, {}), ('pin-success', N, {}), ('notification-expires', N, {}), ('unpin-expired', N, {})], N, {}, 'pinned'), {'phase': 'gone', 'value': ['other']})
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 unpin-success event leaves a notification in pinned instead of unpinned.

Inbox notification pin persistence is a bounded visual-notification workflow with mutable policy state {'pinned': ['other'], 'item': 'n', 'capacity': 3, 'expired': False, 'retained': True}. Its default transition relation is {('unpinned', 'pin-click'): 'pin-pending', ('pin-pending', 'pin-success'): 'pinned', ('pin-pending', 'pin-failure'): 'pin-error', ('pinned', 'unpin-click'): 'unpin-pending', ('unpin-pending', 'unpin-success'): 'unpinned', ('unpin-pending', 'unpin-failure'): 'unpin-error', ('pin-error', 'cancel-pin-intent'): 'unpinned', ('unpin-error', 'cancel-unpin-intent'): 'pinned', ('pinned', 'notification-expires'): 'expired-pinned', ('expired-pinned', 'unpin-expired'): 'gone'}; 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