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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'older-appended', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'end-of-history', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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 head-anchor-ready event leaves a notification in older-appended instead of first-page.

Inbox history pagination presentation is a bounded visual-notification workflow with mutable policy state {'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True}. Its default transition relation is {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}; 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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'older-appended', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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)

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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'end-of-history', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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 filter-changed event leaves a notification in older-appended instead of new-filter-loading.

Inbox history pagination presentation is a bounded visual-notification workflow with mutable policy state {'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True}. Its default transition relation is {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}; 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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'older-appended'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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 = {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'first-page'}
    data = json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True})) if payload is None else {**json.loads(json.dumps({'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': 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 in ('nonempty-page','empty-page','old-page-arrives') and arg.get('filter',data['filter']) != data['filter']: event = 'unknown-event'
            if event == 'nonempty-page': data['rows'] += [x for x in arg.get('rows',['older']) if x not in data['rows']]
            if event == 'filter-changed': data['filter'] += 1
            state = table.get((state, event), state)
            if state == 'end-of-history': data['has_more'] = False
            if state == 'first-page': data['anchor'] = data['rows'][0] if data['rows'] else None
            if state == 'new-filter-loading': 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('first-page / load-older', solve('first-page', [('load-older', N)], N), 'loading-older')
check('loading-older / nonempty-page', solve('loading-older', [('nonempty-page', N)], N), 'older-appended')
check('loading-older / empty-page', solve('loading-older', [('empty-page', N)], N), 'end-of-history')
check('loading-older / page-failure', solve('loading-older', [('page-failure', N)], N), 'older-error')
check('older-error / retry-older', solve('older-error', [('retry-older', N)], N), 'loading-older')
check('older-appended / new-head-arrives', solve('older-appended', [('new-head-arrives', N)], N), 'head-update-pending')
check('head-update-pending / show-new-head', solve('head-update-pending', [('show-new-head', N)], N), 'head-restoring')
check('head-restoring / head-anchor-ready', solve('head-restoring', [('head-anchor-ready', N)], N), 'first-page')
check('loading-older / filter-changed', solve('loading-older', [('filter-changed', N)], N), 'new-filter-loading')
check('new-filter-loading / old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N)], N), 'new-filter-loading')
check('older notification incarnation', solve('first-page', [('load-older', N - 1)], N), 'first-page')
check('future notification incarnation', solve('first-page', [('load-older', N + 1)], N), 'first-page')
check('unknown event is inert', solve('first-page', [('unknown-event', N)], N), 'first-page')
check('empty delivery batch', solve('first-page', [], N), 'first-page')
check('N stale deliveries before current delivery', solve('first-page', [('load-older', N - 1)] * N + [('load-older', N)], N), 'loading-older')
check('trace load-older then nonempty-page', solve('first-page', [('load-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace load-older then empty-page', solve('first-page', [('load-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace load-older then page-failure', solve('first-page', [('load-older', N), ('page-failure', N)], N), 'older-error')
check('trace load-older then filter-changed', solve('first-page', [('load-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace nonempty-page then new-head-arrives', solve('loading-older', [('nonempty-page', N), ('new-head-arrives', N)], N), 'head-update-pending')
check('trace page-failure then retry-older', solve('loading-older', [('page-failure', N), ('retry-older', N)], N), 'loading-older')
check('trace retry-older then nonempty-page', solve('older-error', [('retry-older', N), ('nonempty-page', N)], N), 'older-appended')
check('trace retry-older then empty-page', solve('older-error', [('retry-older', N), ('empty-page', N)], N), 'end-of-history')
check('trace retry-older then page-failure', solve('older-error', [('retry-older', N), ('page-failure', N)], N), 'older-error')
check('trace retry-older then filter-changed', solve('older-error', [('retry-older', N), ('filter-changed', N)], N), 'new-filter-loading')
check('trace new-head-arrives then show-new-head', solve('older-appended', [('new-head-arrives', N), ('show-new-head', N)], N), 'head-restoring')
check('trace show-new-head then head-anchor-ready', solve('head-update-pending', [('show-new-head', N), ('head-anchor-ready', N)], N), 'first-page')
check('trace head-anchor-ready then load-older', solve('head-restoring', [('head-anchor-ready', N), ('load-older', N)], N), 'loading-older')
check('trace filter-changed then old-page-arrives', solve('loading-older', [('filter-changed', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('trace old-page-arrives then old-page-arrives', solve('new-filter-loading', [('old-page-arrives', N), ('old-page-arrives', N)], N), 'new-filter-loading')
check('domain state regression 1: rows', solve('loading-older', [('nonempty-page', N, {'filter': 1, 'rows': ['stale']})], N, {}, 'rows'), {'phase': 'loading-older', 'value': ['head']})
check('domain state regression 2: rows', solve('loading-older', [('nonempty-page', N, {'rows': ['head', 'old']})], N, {}, 'rows'), {'phase': 'older-appended', 'value': ['head', 'old']})
check('domain state regression 3: has_more', solve('loading-older', [('empty-page', N, {})], N, {}, 'has_more'), {'phase': 'end-of-history', 'value': False})
check('domain state regression 4: rows', solve('loading-older', [('filter-changed', N, {})], N, {}, 'rows'), {'phase': 'new-filter-loading', 'value': []})
check('domain state regression 5: filter', solve('loading-older', [('filter-changed', N, {})], N, {}, 'filter'), {'phase': 'new-filter-loading', 'value': 3})
check('domain state regression 6: anchor', solve('older-appended', [('new-head-arrives', N, {})], N, {'anchor': 'old'}, 'anchor'), {'phase': 'head-update-pending', 'value': 'old'})
check('domain state regression 7: rows', solve('first-page', [('load-older', N, {}), ('filter-changed', N, {}), ('old-page-arrives', N, {'filter': 2, 'rows': ['old']})], N, {}, 'rows'), {'phase': 'new-filter-loading', '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 old-page-arrives event leaves a notification in older-appended instead of new-filter-loading.

Inbox history pagination presentation is a bounded visual-notification workflow with mutable policy state {'rows': ['head'], 'filter': 2, 'anchor': 'head', 'has_more': True}. Its default transition relation is {('first-page', 'load-older'): 'loading-older', ('loading-older', 'nonempty-page'): 'older-appended', ('loading-older', 'empty-page'): 'end-of-history', ('loading-older', 'page-failure'): 'older-error', ('older-error', 'retry-older'): 'loading-older', ('older-appended', 'new-head-arrives'): 'head-update-pending', ('head-update-pending', 'show-new-head'): 'head-restoring', ('head-restoring', 'head-anchor-ready'): 'first-page', ('loading-older', 'filter-changed'): 'new-filter-loading', ('new-filter-loading', 'old-page-arrives'): 'new-filter-loading'}; 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 = {('offscreen', 'enters-viewport'): 'counted', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 = {('offscreen', 'enters-viewport'): 'offscreen', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 enters-viewport event leaves a notification in counted instead of qualifying.

Notification visual impression qualification is a bounded visual-notification workflow with mutable policy state {'exposure': 0, 'required': 4, 'covered': False, 'reports': 0}. Its default transition relation is {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}; 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 = {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'counted', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 = {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'offscreen', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 exposure-satisfied event leaves a notification in counted instead of reporting.

Notification visual impression qualification is a bounded visual-notification workflow with mutable policy state {'exposure': 0, 'required': 4, 'covered': False, 'reports': 0}. Its default transition relation is {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}; 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 = {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'reporting', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 = {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'counted', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}
    data = json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})) if payload is None else {**json.loads(json.dumps({'exposure': 0, 'required': 4, 'covered': False, 'reports': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event == 'covered-by-overlay': data['covered'] = True
            if event == 'overlay-removed':
                data['covered'] = False
                data['exposure'] = 0
            if event == 'exposure-satisfied':
                if not data['covered']: data['exposure'] += max(0,arg.get('elapsed',data['required']))
                if data['exposure'] < data['required'] or data['covered']: event = 'unknown-event'
            if event == 'leaves-viewport' and state == 'qualifying': data['exposure'] = 0
            state = table.get((state, event), state)
            if state == 'counted' and previous != 'counted': data['reports'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('offscreen / enters-viewport', solve('offscreen', [('enters-viewport', N)], N), 'qualifying')
check('qualifying / exposure-satisfied', solve('qualifying', [('exposure-satisfied', N)], N), 'reporting')
check('qualifying / leaves-viewport', solve('qualifying', [('leaves-viewport', N)], N), 'offscreen')
check('reporting / report-success', solve('reporting', [('report-success', N)], N), 'counted')
check('reporting / report-failure', solve('reporting', [('report-failure', N)], N), 'report-error')
check('report-error / retry-report', solve('report-error', [('retry-report', N)], N), 'reporting')
check('qualifying / covered-by-overlay', solve('qualifying', [('covered-by-overlay', N)], N), 'occluded')
check('occluded / overlay-removed', solve('occluded', [('overlay-removed', N)], N), 'qualifying')
check('counted / reenter-viewport', solve('counted', [('reenter-viewport', N)], N), 'counted')
check('reporting / leaves-viewport', solve('reporting', [('leaves-viewport', N)], N), 'reporting')
check('older notification incarnation', solve('offscreen', [('enters-viewport', N - 1)], N), 'offscreen')
check('future notification incarnation', solve('offscreen', [('enters-viewport', N + 1)], N), 'offscreen')
check('unknown event is inert', solve('offscreen', [('unknown-event', N)], N), 'offscreen')
check('empty delivery batch', solve('offscreen', [], N), 'offscreen')
check('N stale deliveries before current delivery', solve('offscreen', [('enters-viewport', N - 1)] * N + [('enters-viewport', N)], N), 'qualifying')
check('trace enters-viewport then exposure-satisfied', solve('offscreen', [('enters-viewport', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace enters-viewport then leaves-viewport', solve('offscreen', [('enters-viewport', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace enters-viewport then covered-by-overlay', solve('offscreen', [('enters-viewport', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace exposure-satisfied then report-success', solve('qualifying', [('exposure-satisfied', N), ('report-success', N)], N), 'counted')
check('trace exposure-satisfied then report-failure', solve('qualifying', [('exposure-satisfied', N), ('report-failure', N)], N), 'report-error')
check('trace exposure-satisfied then leaves-viewport', solve('qualifying', [('exposure-satisfied', N), ('leaves-viewport', N)], N), 'reporting')
check('trace leaves-viewport then enters-viewport', solve('qualifying', [('leaves-viewport', N), ('enters-viewport', N)], N), 'qualifying')
check('trace report-success then reenter-viewport', solve('reporting', [('report-success', N), ('reenter-viewport', N)], N), 'counted')
check('trace report-failure then retry-report', solve('reporting', [('report-failure', N), ('retry-report', N)], N), 'reporting')
check('trace retry-report then report-success', solve('report-error', [('retry-report', N), ('report-success', N)], N), 'counted')
check('trace retry-report then report-failure', solve('report-error', [('retry-report', N), ('report-failure', N)], N), 'report-error')
check('trace retry-report then leaves-viewport', solve('report-error', [('retry-report', N), ('leaves-viewport', N)], N), 'reporting')
check('trace covered-by-overlay then overlay-removed', solve('qualifying', [('covered-by-overlay', N), ('overlay-removed', N)], N), 'qualifying')
check('trace overlay-removed then exposure-satisfied', solve('occluded', [('overlay-removed', N), ('exposure-satisfied', N)], N), 'reporting')
check('trace overlay-removed then leaves-viewport', solve('occluded', [('overlay-removed', N), ('leaves-viewport', N)], N), 'offscreen')
check('trace overlay-removed then covered-by-overlay', solve('occluded', [('overlay-removed', N), ('covered-by-overlay', N)], N), 'occluded')
check('trace reenter-viewport then reenter-viewport', solve('counted', [('reenter-viewport', N), ('reenter-viewport', N)], N), 'counted')
check('trace leaves-viewport then report-success', solve('reporting', [('leaves-viewport', N), ('report-success', N)], N), 'counted')
check('trace leaves-viewport then report-failure', solve('reporting', [('leaves-viewport', N), ('report-failure', N)], N), 'report-error')
check('trace leaves-viewport then leaves-viewport', solve('reporting', [('leaves-viewport', N), ('leaves-viewport', N)], N), 'reporting')
check('domain state regression 1: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
check('domain state regression 2: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 4})], N, {}, 'exposure'), {'phase': 'reporting', 'value': 4})
check('domain state regression 3: exposure', solve('qualifying', [('exposure-satisfied', N, {'elapsed': 9})], N, {'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 4: exposure', solve('occluded', [('overlay-removed', N, {})], N, {'exposure': 3, 'covered': True}, 'exposure'), {'phase': 'qualifying', 'value': 0})
check('domain state regression 5: exposure', solve('qualifying', [('leaves-viewport', N, {})], N, {'exposure': 3}, 'exposure'), {'phase': 'offscreen', 'value': 0})
check('domain state regression 6: reports', solve('counted', [('reenter-viewport', N, {})], N, {'reports': 1}, 'reports'), {'phase': 'counted', 'value': 1})
check('domain state regression 7: exposure', solve('offscreen', [('enters-viewport', N, {}), ('exposure-satisfied', N, {'elapsed': 3}), ('covered-by-overlay', N, {}), ('overlay-removed', N, {}), ('exposure-satisfied', N, {'elapsed': 3})], N, {}, 'exposure'), {'phase': 'qualifying', 'value': 3})
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 leaves-viewport event leaves a notification in reporting instead of offscreen.

Notification visual impression qualification is a bounded visual-notification workflow with mutable policy state {'exposure': 0, 'required': 4, 'covered': False, 'reports': 0}. Its default transition relation is {('offscreen', 'enters-viewport'): 'qualifying', ('qualifying', 'exposure-satisfied'): 'reporting', ('qualifying', 'leaves-viewport'): 'offscreen', ('reporting', 'report-success'): 'counted', ('reporting', 'report-failure'): 'report-error', ('report-error', 'retry-report'): 'reporting', ('qualifying', 'covered-by-overlay'): 'occluded', ('occluded', 'overlay-removed'): 'qualifying', ('counted', 'reenter-viewport'): 'counted', ('reporting', 'leaves-viewport'): 'reporting'}; 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