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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduled', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'visible', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 confirm-future event leaves a notification in scheduled instead of scheduling.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'visible', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduling', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 persisted event leaves a notification in visible instead of scheduled.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'scheduled', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'hidden', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 persist-failed event leaves a notification in scheduled instead of snooze-error.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'hidden', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'scheduling', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 wake-time event leaves a notification in hidden instead of resurfaced.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'visible', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'hidden', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 cancel-snooze event leaves a notification in visible instead of cancel-pending.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'scheduled', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 = {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'hidden', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}
    data = json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})) if payload is None else {**json.loads(json.dumps({'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'], arg.get('now',data['now']))
            if event == 'confirm-future':
                wake = arg.get('wake',data['wake'])
                if wake <= data['now']: event = 'unknown-event'
                else: data['wake'] = wake
            if event == 'wake-time' and data['now'] < data['wake'] and 'now' in arg: event = 'unknown-event'
            state = table.get((state, event), state)
            if state == 'scheduled': data['scheduled'] = True
            if state in ('visible','resurfaced'): data['scheduled'] = False
            if state == 'resurfaced' and previous != 'resurfaced': data['resurfaced'] += 1
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('visible / open-snooze', solve('visible', [('open-snooze', N)], N), 'choosing')
check('choosing / cancel-chooser', solve('choosing', [('cancel-chooser', N)], N), 'visible')
check('choosing / confirm-future', solve('choosing', [('confirm-future', N)], N), 'scheduling')
check('scheduling / persisted', solve('scheduling', [('persisted', N)], N), 'scheduled')
check('scheduling / persist-failed', solve('scheduling', [('persist-failed', N)], N), 'snooze-error')
check('scheduled / wake-time', solve('scheduled', [('wake-time', N)], N), 'resurfaced')
check('scheduled / cancel-snooze', solve('scheduled', [('cancel-snooze', N)], N), 'cancel-pending')
check('cancel-pending / cancel-accepted', solve('cancel-pending', [('cancel-accepted', N)], N), 'visible')
check('resurfaced / wake-replay', solve('resurfaced', [('wake-replay', N)], N), 'resurfaced')
check('snooze-error / choose-again', solve('snooze-error', [('choose-again', N)], N), 'choosing')
check('older notification incarnation', solve('visible', [('open-snooze', N - 1)], N), 'visible')
check('future notification incarnation', solve('visible', [('open-snooze', N + 1)], N), 'visible')
check('unknown event is inert', solve('visible', [('unknown-event', N)], N), 'visible')
check('empty delivery batch', solve('visible', [], N), 'visible')
check('N stale deliveries before current delivery', solve('visible', [('open-snooze', N - 1)] * N + [('open-snooze', N)], N), 'choosing')
check('trace open-snooze then cancel-chooser', solve('visible', [('open-snooze', N), ('cancel-chooser', N)], N), 'visible')
check('trace open-snooze then confirm-future', solve('visible', [('open-snooze', N), ('confirm-future', N)], N), 'scheduling')
check('trace cancel-chooser then open-snooze', solve('choosing', [('cancel-chooser', N), ('open-snooze', N)], N), 'choosing')
check('trace confirm-future then persisted', solve('choosing', [('confirm-future', N), ('persisted', N)], N), 'scheduled')
check('trace confirm-future then persist-failed', solve('choosing', [('confirm-future', N), ('persist-failed', N)], N), 'snooze-error')
check('trace persisted then wake-time', solve('scheduling', [('persisted', N), ('wake-time', N)], N), 'resurfaced')
check('trace persisted then cancel-snooze', solve('scheduling', [('persisted', N), ('cancel-snooze', N)], N), 'cancel-pending')
check('trace persist-failed then choose-again', solve('scheduling', [('persist-failed', N), ('choose-again', N)], N), 'choosing')
check('trace wake-time then wake-replay', solve('scheduled', [('wake-time', N), ('wake-replay', N)], N), 'resurfaced')
check('trace cancel-snooze then cancel-accepted', solve('scheduled', [('cancel-snooze', N), ('cancel-accepted', N)], N), 'visible')
check('trace cancel-accepted then open-snooze', solve('cancel-pending', [('cancel-accepted', N), ('open-snooze', N)], N), 'choosing')
check('trace wake-replay then wake-replay', solve('resurfaced', [('wake-replay', N), ('wake-replay', N)], N), 'resurfaced')
check('trace choose-again then cancel-chooser', solve('snooze-error', [('choose-again', N), ('cancel-chooser', N)], N), 'visible')
check('trace choose-again then confirm-future', solve('snooze-error', [('choose-again', N), ('confirm-future', N)], N), 'scheduling')
check('domain state regression 1: wake', solve('choosing', [('confirm-future', N, {'wake': 10})], N, {}, 'wake'), {'phase': 'choosing', 'value': 30})
check('domain state regression 2: wake', solve('choosing', [('confirm-future', N, {'wake': 50})], N, {}, 'wake'), {'phase': 'scheduling', 'value': 50})
check('domain state regression 3: scheduled', solve('scheduled', [('wake-time', N, {'now': 29})], N, {'scheduled': True}, 'scheduled'), {'phase': 'scheduled', 'value': True})
check('domain state regression 4: resurfaced', solve('scheduled', [('wake-time', N, {'now': 30})], N, {'scheduled': True}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 5: resurfaced', solve('resurfaced', [('wake-replay', N, {})], N, {'resurfaced': 1}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
check('domain state regression 6: scheduled', solve('cancel-pending', [('cancel-accepted', N, {})], N, {'scheduled': True}, 'scheduled'), {'phase': 'visible', 'value': False})
check('domain state regression 7: resurfaced', solve('choosing', [('confirm-future', N, {'wake': 40}), ('persisted', N, {}), ('wake-time', N, {'now': 39}), ('wake-time', N, {'now': 40}), ('wake-replay', N, {})], N, {}, 'resurfaced'), {'phase': 'resurfaced', 'value': 1})
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 cancel-accepted event leaves a notification in scheduled instead of visible.

Notification snooze presentation is a bounded visual-notification workflow with mutable policy state {'now': 10, 'wake': 30, 'scheduled': False, 'resurfaced': 0}. Its default transition relation is {('visible', 'open-snooze'): 'choosing', ('choosing', 'cancel-chooser'): 'visible', ('choosing', 'confirm-future'): 'scheduling', ('scheduling', 'persisted'): 'scheduled', ('scheduling', 'persist-failed'): 'snooze-error', ('scheduled', 'wake-time'): 'resurfaced', ('scheduled', 'cancel-snooze'): 'cancel-pending', ('cancel-pending', 'cancel-accepted'): 'visible', ('resurfaced', 'wake-replay'): 'resurfaced', ('snooze-error', 'choose-again'): 'choosing'}; 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