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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'empty', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'empty'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'collecting', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'empty'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 collapse-digest event leaves a notification in empty instead of published.

Notification digest assembly presentation is a bounded visual-notification workflow with mutable policy state {'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []}. Its default transition relation is {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'empty'}; 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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'collecting', ('collecting', 'close-empty-window'): 'empty'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'published', ('collecting', 'close-empty-window'): 'empty'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 late-item event leaves a notification in collecting instead of sealing.

Notification digest assembly presentation is a bounded visual-notification workflow with mutable policy state {'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []}. Its default transition relation is {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'empty'}; 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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'published'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 = {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'ready'}
    data = json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})) if payload is None else {**json.loads(json.dumps({'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []})), **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 == 'close-empty-window' and data['members']: event = 'close-window'
            if event == 'late-item': data['next_window'].append(arg.get('item','late'))
            if event == 'seal-success': data['sealed'] = data['members'][:]
            state = table.get((state, event), state)
            if state == 'published': data['shown'] = data['sealed'][:]
            if state == 'empty': data['shown'] = []
    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('collecting / close-window', solve('collecting', [('close-window', N)], N), 'sealing')
check('sealing / seal-success', solve('sealing', [('seal-success', N)], N), 'ready')
check('ready / show-digest', solve('ready', [('show-digest', N)], N), 'publishing')
check('publishing / paint-success', solve('publishing', [('paint-success', N)], N), 'published')
check('publishing / paint-failure', solve('publishing', [('paint-failure', N)], N), 'publish-error')
check('publish-error / retry-show', solve('publish-error', [('retry-show', N)], N), 'publishing')
check('published / expand-digest', solve('published', [('expand-digest', N)], N), 'expanded')
check('expanded / collapse-digest', solve('expanded', [('collapse-digest', N)], N), 'published')
check('sealing / late-item', solve('sealing', [('late-item', N)], N), 'sealing')
check('collecting / close-empty-window', solve('collecting', [('close-empty-window', N)], N, {'members': []}), 'empty')
check('older notification incarnation', solve('collecting', [('close-window', N - 1)], N), 'collecting')
check('future notification incarnation', solve('collecting', [('close-window', N + 1)], N), 'collecting')
check('unknown event is inert', solve('collecting', [('unknown-event', N)], N), 'collecting')
check('empty delivery batch', solve('collecting', [], N), 'collecting')
check('N stale deliveries before current delivery', solve('collecting', [('close-window', N - 1)] * N + [('close-window', N)], N), 'sealing')
check('trace close-window then seal-success', solve('collecting', [('close-window', N), ('seal-success', N)], N), 'ready')
check('trace close-window then late-item', solve('collecting', [('close-window', N), ('late-item', N)], N), 'sealing')
check('trace seal-success then show-digest', solve('sealing', [('seal-success', N), ('show-digest', N)], N), 'publishing')
check('trace show-digest then paint-success', solve('ready', [('show-digest', N), ('paint-success', N)], N), 'published')
check('trace show-digest then paint-failure', solve('ready', [('show-digest', N), ('paint-failure', N)], N), 'publish-error')
check('trace paint-success then expand-digest', solve('publishing', [('paint-success', N), ('expand-digest', N)], N), 'expanded')
check('trace paint-failure then retry-show', solve('publishing', [('paint-failure', N), ('retry-show', N)], N), 'publishing')
check('trace retry-show then paint-success', solve('publish-error', [('retry-show', N), ('paint-success', N)], N), 'published')
check('trace retry-show then paint-failure', solve('publish-error', [('retry-show', N), ('paint-failure', N)], N), 'publish-error')
check('trace expand-digest then collapse-digest', solve('published', [('expand-digest', N), ('collapse-digest', N)], N), 'published')
check('trace collapse-digest then expand-digest', solve('expanded', [('collapse-digest', N), ('expand-digest', N)], N), 'expanded')
check('trace late-item then seal-success', solve('sealing', [('late-item', N), ('seal-success', N)], N), 'ready')
check('trace late-item then late-item', solve('sealing', [('late-item', N), ('late-item', N)], N), 'sealing')
check('domain state regression 1: members', solve('collecting', [('close-empty-window', N, {})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 2: shown', solve('collecting', [('close-empty-window', N, {})], N, {'members': []}, 'shown'), {'phase': 'empty', 'value': []})
check('domain state regression 3: sealed', solve('sealing', [('seal-success', N, {})], N, {}, 'sealed'), {'phase': 'ready', 'value': ['a', 'b']})
check('domain state regression 4: members', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'members'), {'phase': 'sealing', 'value': ['a', 'b']})
check('domain state regression 5: next_window', solve('sealing', [('late-item', N, {'item': 'c'})], N, {}, 'next_window'), {'phase': 'sealing', 'value': ['c']})
check('domain state regression 6: shown', solve('publishing', [('paint-success', N, {})], N, {'sealed': ['a']}, 'shown'), {'phase': 'published', 'value': ['a']})
check('domain state regression 7: shown', solve('collecting', [('close-window', N, {}), ('late-item', N, {'item': 'c'}), ('seal-success', N, {}), ('show-digest', N, {}), ('paint-success', N, {})], N, {}, 'shown'), {'phase': 'published', 'value': ['a', 'b']})
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 close-empty-window event leaves a notification in published instead of empty.

Notification digest assembly presentation is a bounded visual-notification workflow with mutable policy state {'members': ['a', 'b'], 'sealed': [], 'next_window': [], 'shown': []}. Its default transition relation is {('collecting', 'close-window'): 'sealing', ('sealing', 'seal-success'): 'ready', ('ready', 'show-digest'): 'publishing', ('publishing', 'paint-success'): 'published', ('publishing', 'paint-failure'): 'publish-error', ('publish-error', 'retry-show'): 'publishing', ('published', 'expand-digest'): 'expanded', ('expanded', 'collapse-digest'): 'published', ('sealing', 'late-item'): 'sealing', ('collecting', 'close-empty-window'): 'empty'}; 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 = {('thumbnail', 'preview-click'): 'open', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 = {('thumbnail', 'preview-click'): 'failed', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 preview-click event leaves a notification in open instead of fetching.

Notification attachment preview is a bounded visual-notification workflow with mutable policy state {'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []}. Its default transition relation is {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'open', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'thumbnail', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 content-ready event leaves a notification in open instead of decoding.

Notification attachment preview is a bounded visual-notification workflow with mutable policy state {'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []}. Its default transition relation is {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'thumbnail', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'fetching', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 decode-success event leaves a notification in thumbnail instead of open.

Notification attachment preview is a bounded visual-notification workflow with mutable policy state {'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []}. Its default transition relation is {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access


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

N = 1
observations = []
def solve(initial, events, generation, payload=None, observe=None):
    table = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'open', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 = {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'fetch-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}
    data = json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})) if payload is None else {**json.loads(json.dumps({'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []})), **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 ('content-ready','decode-success','decode-failure') and arg.get('attachment',data['attachment']) != data['attachment']: event = 'unknown-event'
            if event == 'content-ready': data['bytes'] = arg.get('bytes',12)
            if event == 'decode-success' and data['bytes'] == 0 and arg.get('require_bytes',False): event = 'decode-failure'
            state = table.get((state, event), state)
            if state == 'open': data['decoded'] = True
            if state in ('thumbnail','canceled'): data['decoded'] = False
            if state == 'download-pending': data['download'].append(data['attachment'])
    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('thumbnail / preview-click', solve('thumbnail', [('preview-click', N)], N), 'fetching')
check('fetching / content-ready', solve('fetching', [('content-ready', N)], N), 'decoding')
check('decoding / decode-success', solve('decoding', [('decode-success', N)], N), 'open')
check('decoding / decode-failure', solve('decoding', [('decode-failure', N)], N), 'decode-error')
check('fetching / fetch-failure', solve('fetching', [('fetch-failure', N)], N), 'fetch-error')
check('fetch-error / retry-fetch', solve('fetch-error', [('retry-fetch', N)], N), 'fetching')
check('open / close-preview', solve('open', [('close-preview', N)], N), 'thumbnail')
check('fetching / close-preview', solve('fetching', [('close-preview', N)], N), 'canceled')
check('canceled / late-content-ready', solve('canceled', [('late-content-ready', N)], N), 'thumbnail')
check('decode-error / download-original', solve('decode-error', [('download-original', N)], N), 'download-pending')
check('older notification incarnation', solve('thumbnail', [('preview-click', N - 1)], N), 'thumbnail')
check('future notification incarnation', solve('thumbnail', [('preview-click', N + 1)], N), 'thumbnail')
check('unknown event is inert', solve('thumbnail', [('unknown-event', N)], N), 'thumbnail')
check('empty delivery batch', solve('thumbnail', [], N), 'thumbnail')
check('N stale deliveries before current delivery', solve('thumbnail', [('preview-click', N - 1)] * N + [('preview-click', N)], N), 'fetching')
check('trace preview-click then content-ready', solve('thumbnail', [('preview-click', N), ('content-ready', N)], N), 'decoding')
check('trace preview-click then fetch-failure', solve('thumbnail', [('preview-click', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace preview-click then close-preview', solve('thumbnail', [('preview-click', N), ('close-preview', N)], N), 'canceled')
check('trace content-ready then decode-success', solve('fetching', [('content-ready', N), ('decode-success', N)], N), 'open')
check('trace content-ready then decode-failure', solve('fetching', [('content-ready', N), ('decode-failure', N)], N), 'decode-error')
check('trace decode-success then close-preview', solve('decoding', [('decode-success', N), ('close-preview', N)], N), 'thumbnail')
check('trace decode-failure then download-original', solve('decoding', [('decode-failure', N), ('download-original', N)], N), 'download-pending')
check('trace fetch-failure then retry-fetch', solve('fetching', [('fetch-failure', N), ('retry-fetch', N)], N), 'fetching')
check('trace retry-fetch then content-ready', solve('fetch-error', [('retry-fetch', N), ('content-ready', N)], N), 'decoding')
check('trace retry-fetch then fetch-failure', solve('fetch-error', [('retry-fetch', N), ('fetch-failure', N)], N), 'fetch-error')
check('trace retry-fetch then close-preview', solve('fetch-error', [('retry-fetch', N), ('close-preview', N)], N), 'canceled')
check('trace close-preview then preview-click', solve('open', [('close-preview', N), ('preview-click', N)], N), 'fetching')
check('trace close-preview then late-content-ready', solve('fetching', [('close-preview', N), ('late-content-ready', N)], N), 'thumbnail')
check('trace late-content-ready then preview-click', solve('canceled', [('late-content-ready', N), ('preview-click', N)], N), 'fetching')
check('domain state regression 1: bytes', solve('fetching', [('content-ready', N, {'attachment': 'old', 'bytes': 7})], N, {}, 'bytes'), {'phase': 'fetching', 'value': 0})
check('domain state regression 2: bytes', solve('fetching', [('content-ready', N, {'bytes': 7})], N, {}, 'bytes'), {'phase': 'decoding', 'value': 7})
check('domain state regression 3: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {}, 'decoded'), {'phase': 'decode-error', 'value': False})
check('domain state regression 4: decoded', solve('decoding', [('decode-success', N, {'require_bytes': True})], N, {'bytes': 7}, 'decoded'), {'phase': 'open', 'value': True})
check('domain state regression 5: download', solve('decode-error', [('download-original', N, {})], N, {}, 'download'), {'phase': 'download-pending', 'value': ['photo']})
check('domain state regression 6: decoded', solve('open', [('close-preview', N, {})], N, {'decoded': True}, 'decoded'), {'phase': 'thumbnail', 'value': False})
check('domain state regression 7: decoded', solve('thumbnail', [('preview-click', N, {}), ('content-ready', N, {'bytes': 8}), ('decode-success', N, {'require_bytes': True}), ('close-preview', N, {})], N, {}, 'decoded'), {'phase': 'thumbnail', 'value': False})
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 decode-failure event leaves a notification in open instead of decode-error.

Notification attachment preview is a bounded visual-notification workflow with mutable policy state {'attachment': 'photo', 'bytes': 0, 'decoded': False, 'download': []}. Its default transition relation is {('thumbnail', 'preview-click'): 'fetching', ('fetching', 'content-ready'): 'decoding', ('decoding', 'decode-success'): 'open', ('decoding', 'decode-failure'): 'decode-error', ('fetching', 'fetch-failure'): 'fetch-error', ('fetch-error', 'retry-fetch'): 'fetching', ('open', 'close-preview'): 'thumbnail', ('fetching', 'close-preview'): 'canceled', ('canceled', 'late-content-ready'): 'thumbnail', ('decode-error', 'download-original'): 'download-pending'}; domain inputs can suppress or redirect transitions and update policy fields, as specified in solve. Different-generation events are inert. Batches apply in order. Optional observe returns the selected policy field alongside the phase.

split: open-access