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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'succeeded', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'submitting', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 rejection event leaves a notification in succeeded instead of failed.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'running', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'succeeded', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The retry-click event leaves a notification in running instead of submitting.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'failed', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'ready', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 transport-unknown event leaves a notification in failed instead of uncertain.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'submitting', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'succeeded', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 status-confirmed-running event leaves a notification in submitting instead of running.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'succeeded', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'running', ('succeeded', 'stale-rejection'): 'succeeded'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 status-confirmed-absent event leaves a notification in succeeded instead of ready.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'failed'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 = {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'ready'}
    data = json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})) if payload is None else {**json.loads(json.dumps({'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            command = arg.get('command', data['command'])
            if event in ('accepted', 'completion', 'rejection') and command != data['command']: event = 'unknown-event'
            if event in ('action-click','retry-click') and data['command'] not in data['pending']: data['pending'].append(data['command'])
            if event == 'transport-unknown': data['transport_known'] = False
            if event.startswith('status-confirmed-'): data['transport_known'] = True
            state = table.get((state, event), state)
            if state == 'succeeded':
                data['pending'] = [x for x in data['pending'] if x != data['command']]
                if data['command'] not in data['completed']: data['completed'].append(data['command'])
            if state == 'ready': data['pending'] = [x for x in data['pending'] if x != data['command']]
    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('ready / action-click', solve('ready', [('action-click', N)], N), 'submitting')
check('submitting / second-click', solve('submitting', [('second-click', N)], N), 'submitting')
check('submitting / accepted', solve('submitting', [('accepted', N)], N), 'running')
check('running / completion', solve('running', [('completion', N)], N), 'succeeded')
check('running / rejection', solve('running', [('rejection', N)], N), 'failed')
check('failed / retry-click', solve('failed', [('retry-click', N)], N), 'submitting')
check('submitting / transport-unknown', solve('submitting', [('transport-unknown', N)], N), 'uncertain')
check('uncertain / status-confirmed-running', solve('uncertain', [('status-confirmed-running', N)], N), 'running')
check('uncertain / status-confirmed-absent', solve('uncertain', [('status-confirmed-absent', N)], N), 'ready')
check('succeeded / stale-rejection', solve('succeeded', [('stale-rejection', N)], N), 'succeeded')
check('older notification incarnation', solve('ready', [('action-click', N - 1)], N), 'ready')
check('future notification incarnation', solve('ready', [('action-click', N + 1)], N), 'ready')
check('unknown event is inert', solve('ready', [('unknown-event', N)], N), 'ready')
check('empty delivery batch', solve('ready', [], N), 'ready')
check('N stale deliveries before current delivery', solve('ready', [('action-click', N - 1)] * N + [('action-click', N)], N), 'submitting')
check('trace action-click then second-click', solve('ready', [('action-click', N), ('second-click', N)], N), 'submitting')
check('trace action-click then accepted', solve('ready', [('action-click', N), ('accepted', N)], N), 'running')
check('trace action-click then transport-unknown', solve('ready', [('action-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace second-click then second-click', solve('submitting', [('second-click', N), ('second-click', N)], N), 'submitting')
check('trace second-click then accepted', solve('submitting', [('second-click', N), ('accepted', N)], N), 'running')
check('trace second-click then transport-unknown', solve('submitting', [('second-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace accepted then completion', solve('submitting', [('accepted', N), ('completion', N)], N), 'succeeded')
check('trace accepted then rejection', solve('submitting', [('accepted', N), ('rejection', N)], N), 'failed')
check('trace completion then stale-rejection', solve('running', [('completion', N), ('stale-rejection', N)], N), 'succeeded')
check('trace rejection then retry-click', solve('running', [('rejection', N), ('retry-click', N)], N), 'submitting')
check('trace retry-click then second-click', solve('failed', [('retry-click', N), ('second-click', N)], N), 'submitting')
check('trace retry-click then accepted', solve('failed', [('retry-click', N), ('accepted', N)], N), 'running')
check('trace retry-click then transport-unknown', solve('failed', [('retry-click', N), ('transport-unknown', N)], N), 'uncertain')
check('trace transport-unknown then status-confirmed-running', solve('submitting', [('transport-unknown', N), ('status-confirmed-running', N)], N), 'running')
check('trace transport-unknown then status-confirmed-absent', solve('submitting', [('transport-unknown', N), ('status-confirmed-absent', N)], N), 'ready')
check('trace status-confirmed-running then completion', solve('uncertain', [('status-confirmed-running', N), ('completion', N)], N), 'succeeded')
check('trace status-confirmed-running then rejection', solve('uncertain', [('status-confirmed-running', N), ('rejection', N)], N), 'failed')
check('trace status-confirmed-absent then action-click', solve('uncertain', [('status-confirmed-absent', N), ('action-click', N)], N), 'submitting')
check('trace stale-rejection then stale-rejection', solve('succeeded', [('stale-rejection', N), ('stale-rejection', N)], N), 'succeeded')
check('domain state regression 1: completed', solve('running', [('completion', N, {'command': 'old'})], N, {}, 'completed'), {'phase': 'running', 'value': []})
check('domain state regression 2: pending', solve('running', [('completion', N, {'command': 'c1'})], N, {'pending': ['c1']}, 'pending'), {'phase': 'succeeded', 'value': []})
check('domain state regression 3: completed', solve('running', [('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
check('domain state regression 4: transport_known', solve('submitting', [('transport-unknown', N, {})], N, {}, 'transport_known'), {'phase': 'uncertain', 'value': False})
check('domain state regression 5: transport_known', solve('uncertain', [('status-confirmed-running', N, {})], N, {'transport_known': False}, 'transport_known'), {'phase': 'running', 'value': True})
check('domain state regression 6: pending', solve('ready', [('action-click', N, {})], N, {'pending': ['other']}, 'pending'), {'phase': 'submitting', 'value': ['other', 'c1']})
check('domain state regression 7: completed', solve('ready', [('action-click', N, {}), ('accepted', N, {}), ('completion', N, {})], N, {}, 'completed'), {'phase': 'succeeded', 'value': ['c1']})
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 stale-rejection event leaves a notification in failed instead of succeeded.

Toast action request feedback is a bounded visual-notification workflow with mutable policy state {'command': 'c1', 'pending': [], 'completed': [], 'transport_known': True}. Its default transition relation is {('ready', 'action-click'): 'submitting', ('submitting', 'second-click'): 'submitting', ('submitting', 'accepted'): 'running', ('running', 'completion'): 'succeeded', ('running', 'rejection'): 'failed', ('failed', 'retry-click'): 'submitting', ('submitting', 'transport-unknown'): 'uncertain', ('uncertain', 'status-confirmed-running'): 'running', ('uncertain', 'status-confirmed-absent'): 'ready', ('succeeded', 'stale-rejection'): 'succeeded'}; 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