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 = {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'unmute-pending', ('mute-error', 'cancel-mute-intent'): 'audible-channel'}
    data = json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})) if payload is None else {**json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})), **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 == 'mute-window-ended' and data['now'] < data['until'] and 'now' in arg: event = 'unknown-event'
            if event == 'extend-mute': data['extension'] = max(data['until'],arg.get('until',data['extension']))
            if event == 'extension-confirmed': data['until'] = data['extension']
            state = table.get((state, event), state)
            if state in ('muted','muted-extended'): data['muted'] = True
            if state == 'audible-channel': data['muted'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('audible-channel / mute-for-session', solve('audible-channel', [('mute-for-session', N)], N), 'mute-pending')
check('mute-pending / mute-confirmed', solve('mute-pending', [('mute-confirmed', N)], N), 'muted')
check('mute-pending / mute-rejected', solve('mute-pending', [('mute-rejected', N)], N), 'mute-error')
check('muted / mute-window-ended', solve('muted', [('mute-window-ended', N)], N), 'unmute-pending')
check('unmute-pending / unmute-confirmed', solve('unmute-pending', [('unmute-confirmed', N)], N), 'audible-channel')
check('unmute-pending / unmute-rejected', solve('unmute-pending', [('unmute-rejected', N)], N), 'unmute-error')
check('muted / extend-mute', solve('muted', [('extend-mute', N)], N), 'extend-pending')
check('extend-pending / extension-confirmed', solve('extend-pending', [('extension-confirmed', N)], N), 'muted-extended')
check('muted-extended / old-window-ended', solve('muted-extended', [('old-window-ended', N)], N), 'muted-extended')
check('mute-error / cancel-mute-intent', solve('mute-error', [('cancel-mute-intent', N)], N), 'audible-channel')
check('older notification incarnation', solve('audible-channel', [('mute-for-session', N - 1)], N), 'audible-channel')
check('future notification incarnation', solve('audible-channel', [('mute-for-session', N + 1)], N), 'audible-channel')
check('unknown event is inert', solve('audible-channel', [('unknown-event', N)], N), 'audible-channel')
check('empty delivery batch', solve('audible-channel', [], N), 'audible-channel')
check('N stale deliveries before current delivery', solve('audible-channel', [('mute-for-session', N - 1)] * N + [('mute-for-session', N)], N), 'mute-pending')
check('trace mute-for-session then mute-confirmed', solve('audible-channel', [('mute-for-session', N), ('mute-confirmed', N)], N), 'muted')
check('trace mute-for-session then mute-rejected', solve('audible-channel', [('mute-for-session', N), ('mute-rejected', N)], N), 'mute-error')
check('trace mute-confirmed then mute-window-ended', solve('mute-pending', [('mute-confirmed', N), ('mute-window-ended', N)], N), 'unmute-pending')
check('trace mute-confirmed then extend-mute', solve('mute-pending', [('mute-confirmed', N), ('extend-mute', N)], N), 'extend-pending')
check('trace mute-rejected then cancel-mute-intent', solve('mute-pending', [('mute-rejected', N), ('cancel-mute-intent', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-confirmed', solve('muted', [('mute-window-ended', N), ('unmute-confirmed', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-rejected', solve('muted', [('mute-window-ended', N), ('unmute-rejected', N)], N), 'unmute-error')
check('trace unmute-confirmed then mute-for-session', solve('unmute-pending', [('unmute-confirmed', N), ('mute-for-session', N)], N), 'mute-pending')
check('trace extend-mute then extension-confirmed', solve('muted', [('extend-mute', N), ('extension-confirmed', N)], N), 'muted-extended')
check('trace extension-confirmed then old-window-ended', solve('extend-pending', [('extension-confirmed', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace old-window-ended then old-window-ended', solve('muted-extended', [('old-window-ended', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace cancel-mute-intent then mute-for-session', solve('mute-error', [('cancel-mute-intent', N), ('mute-for-session', N)], N), 'mute-pending')
check('domain state regression 1: muted', solve('muted', [('mute-window-ended', N, {'now': 29})], N, {'muted': True}, 'muted'), {'phase': 'muted', 'value': True})
check('domain state regression 2: muted', solve('muted', [('mute-window-ended', N, {'now': 30})], N, {'muted': True}, 'muted'), {'phase': 'unmute-pending', 'value': True})
check('domain state regression 3: extension', solve('muted', [('extend-mute', N, {'until': 20})], N, {}, 'extension'), {'phase': 'extend-pending', 'value': 30})
check('domain state regression 4: until', solve('extend-pending', [('extension-confirmed', N, {})], N, {}, 'until'), {'phase': 'muted-extended', 'value': 60})
check('domain state regression 5: muted', solve('muted-extended', [('old-window-ended', N, {})], N, {'until': 60, 'muted': True}, 'muted'), {'phase': 'muted-extended', 'value': True})
check('domain state regression 6: muted', solve('unmute-pending', [('unmute-confirmed', N, {})], N, {'muted': True}, 'muted'), {'phase': 'audible-channel', 'value': False})
check('domain state regression 7: until', solve('muted', [('extend-mute', N, {'until': 80}), ('extension-confirmed', N, {}), ('old-window-ended', N, {})], N, {'muted': True}, 'until'), {'phase': 'muted-extended', 'value': 80})
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 = {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'audible-channel', ('mute-error', 'cancel-mute-intent'): 'audible-channel'}
    data = json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})) if payload is None else {**json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})), **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 == 'mute-window-ended' and data['now'] < data['until'] and 'now' in arg: event = 'unknown-event'
            if event == 'extend-mute': data['extension'] = max(data['until'],arg.get('until',data['extension']))
            if event == 'extension-confirmed': data['until'] = data['extension']
            state = table.get((state, event), state)
            if state in ('muted','muted-extended'): data['muted'] = True
            if state == 'audible-channel': data['muted'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('audible-channel / mute-for-session', solve('audible-channel', [('mute-for-session', N)], N), 'mute-pending')
check('mute-pending / mute-confirmed', solve('mute-pending', [('mute-confirmed', N)], N), 'muted')
check('mute-pending / mute-rejected', solve('mute-pending', [('mute-rejected', N)], N), 'mute-error')
check('muted / mute-window-ended', solve('muted', [('mute-window-ended', N)], N), 'unmute-pending')
check('unmute-pending / unmute-confirmed', solve('unmute-pending', [('unmute-confirmed', N)], N), 'audible-channel')
check('unmute-pending / unmute-rejected', solve('unmute-pending', [('unmute-rejected', N)], N), 'unmute-error')
check('muted / extend-mute', solve('muted', [('extend-mute', N)], N), 'extend-pending')
check('extend-pending / extension-confirmed', solve('extend-pending', [('extension-confirmed', N)], N), 'muted-extended')
check('muted-extended / old-window-ended', solve('muted-extended', [('old-window-ended', N)], N), 'muted-extended')
check('mute-error / cancel-mute-intent', solve('mute-error', [('cancel-mute-intent', N)], N), 'audible-channel')
check('older notification incarnation', solve('audible-channel', [('mute-for-session', N - 1)], N), 'audible-channel')
check('future notification incarnation', solve('audible-channel', [('mute-for-session', N + 1)], N), 'audible-channel')
check('unknown event is inert', solve('audible-channel', [('unknown-event', N)], N), 'audible-channel')
check('empty delivery batch', solve('audible-channel', [], N), 'audible-channel')
check('N stale deliveries before current delivery', solve('audible-channel', [('mute-for-session', N - 1)] * N + [('mute-for-session', N)], N), 'mute-pending')
check('trace mute-for-session then mute-confirmed', solve('audible-channel', [('mute-for-session', N), ('mute-confirmed', N)], N), 'muted')
check('trace mute-for-session then mute-rejected', solve('audible-channel', [('mute-for-session', N), ('mute-rejected', N)], N), 'mute-error')
check('trace mute-confirmed then mute-window-ended', solve('mute-pending', [('mute-confirmed', N), ('mute-window-ended', N)], N), 'unmute-pending')
check('trace mute-confirmed then extend-mute', solve('mute-pending', [('mute-confirmed', N), ('extend-mute', N)], N), 'extend-pending')
check('trace mute-rejected then cancel-mute-intent', solve('mute-pending', [('mute-rejected', N), ('cancel-mute-intent', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-confirmed', solve('muted', [('mute-window-ended', N), ('unmute-confirmed', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-rejected', solve('muted', [('mute-window-ended', N), ('unmute-rejected', N)], N), 'unmute-error')
check('trace unmute-confirmed then mute-for-session', solve('unmute-pending', [('unmute-confirmed', N), ('mute-for-session', N)], N), 'mute-pending')
check('trace extend-mute then extension-confirmed', solve('muted', [('extend-mute', N), ('extension-confirmed', N)], N), 'muted-extended')
check('trace extension-confirmed then old-window-ended', solve('extend-pending', [('extension-confirmed', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace old-window-ended then old-window-ended', solve('muted-extended', [('old-window-ended', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace cancel-mute-intent then mute-for-session', solve('mute-error', [('cancel-mute-intent', N), ('mute-for-session', N)], N), 'mute-pending')
check('domain state regression 1: muted', solve('muted', [('mute-window-ended', N, {'now': 29})], N, {'muted': True}, 'muted'), {'phase': 'muted', 'value': True})
check('domain state regression 2: muted', solve('muted', [('mute-window-ended', N, {'now': 30})], N, {'muted': True}, 'muted'), {'phase': 'unmute-pending', 'value': True})
check('domain state regression 3: extension', solve('muted', [('extend-mute', N, {'until': 20})], N, {}, 'extension'), {'phase': 'extend-pending', 'value': 30})
check('domain state regression 4: until', solve('extend-pending', [('extension-confirmed', N, {})], N, {}, 'until'), {'phase': 'muted-extended', 'value': 60})
check('domain state regression 5: muted', solve('muted-extended', [('old-window-ended', N, {})], N, {'until': 60, 'muted': True}, 'muted'), {'phase': 'muted-extended', 'value': True})
check('domain state regression 6: muted', solve('unmute-pending', [('unmute-confirmed', N, {})], N, {'muted': True}, 'muted'), {'phase': 'audible-channel', 'value': False})
check('domain state regression 7: until', solve('muted', [('extend-mute', N, {'until': 80}), ('extension-confirmed', N, {}), ('old-window-ended', N, {})], N, {'muted': True}, 'until'), {'phase': 'muted-extended', 'value': 80})
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The old-window-ended event leaves a notification in unmute-pending instead of muted-extended.

Temporary notification mute feedback is a bounded visual-notification workflow with mutable policy state {'until': 30, 'now': 10, 'muted': False, 'extension': 60}. Its default transition relation is {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'muted-extended', ('mute-error', 'cancel-mute-intent'): 'audible-channel'}; 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 = {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'muted-extended', ('mute-error', 'cancel-mute-intent'): 'muted'}
    data = json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})) if payload is None else {**json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})), **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 == 'mute-window-ended' and data['now'] < data['until'] and 'now' in arg: event = 'unknown-event'
            if event == 'extend-mute': data['extension'] = max(data['until'],arg.get('until',data['extension']))
            if event == 'extension-confirmed': data['until'] = data['extension']
            state = table.get((state, event), state)
            if state in ('muted','muted-extended'): data['muted'] = True
            if state == 'audible-channel': data['muted'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('audible-channel / mute-for-session', solve('audible-channel', [('mute-for-session', N)], N), 'mute-pending')
check('mute-pending / mute-confirmed', solve('mute-pending', [('mute-confirmed', N)], N), 'muted')
check('mute-pending / mute-rejected', solve('mute-pending', [('mute-rejected', N)], N), 'mute-error')
check('muted / mute-window-ended', solve('muted', [('mute-window-ended', N)], N), 'unmute-pending')
check('unmute-pending / unmute-confirmed', solve('unmute-pending', [('unmute-confirmed', N)], N), 'audible-channel')
check('unmute-pending / unmute-rejected', solve('unmute-pending', [('unmute-rejected', N)], N), 'unmute-error')
check('muted / extend-mute', solve('muted', [('extend-mute', N)], N), 'extend-pending')
check('extend-pending / extension-confirmed', solve('extend-pending', [('extension-confirmed', N)], N), 'muted-extended')
check('muted-extended / old-window-ended', solve('muted-extended', [('old-window-ended', N)], N), 'muted-extended')
check('mute-error / cancel-mute-intent', solve('mute-error', [('cancel-mute-intent', N)], N), 'audible-channel')
check('older notification incarnation', solve('audible-channel', [('mute-for-session', N - 1)], N), 'audible-channel')
check('future notification incarnation', solve('audible-channel', [('mute-for-session', N + 1)], N), 'audible-channel')
check('unknown event is inert', solve('audible-channel', [('unknown-event', N)], N), 'audible-channel')
check('empty delivery batch', solve('audible-channel', [], N), 'audible-channel')
check('N stale deliveries before current delivery', solve('audible-channel', [('mute-for-session', N - 1)] * N + [('mute-for-session', N)], N), 'mute-pending')
check('trace mute-for-session then mute-confirmed', solve('audible-channel', [('mute-for-session', N), ('mute-confirmed', N)], N), 'muted')
check('trace mute-for-session then mute-rejected', solve('audible-channel', [('mute-for-session', N), ('mute-rejected', N)], N), 'mute-error')
check('trace mute-confirmed then mute-window-ended', solve('mute-pending', [('mute-confirmed', N), ('mute-window-ended', N)], N), 'unmute-pending')
check('trace mute-confirmed then extend-mute', solve('mute-pending', [('mute-confirmed', N), ('extend-mute', N)], N), 'extend-pending')
check('trace mute-rejected then cancel-mute-intent', solve('mute-pending', [('mute-rejected', N), ('cancel-mute-intent', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-confirmed', solve('muted', [('mute-window-ended', N), ('unmute-confirmed', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-rejected', solve('muted', [('mute-window-ended', N), ('unmute-rejected', N)], N), 'unmute-error')
check('trace unmute-confirmed then mute-for-session', solve('unmute-pending', [('unmute-confirmed', N), ('mute-for-session', N)], N), 'mute-pending')
check('trace extend-mute then extension-confirmed', solve('muted', [('extend-mute', N), ('extension-confirmed', N)], N), 'muted-extended')
check('trace extension-confirmed then old-window-ended', solve('extend-pending', [('extension-confirmed', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace old-window-ended then old-window-ended', solve('muted-extended', [('old-window-ended', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace cancel-mute-intent then mute-for-session', solve('mute-error', [('cancel-mute-intent', N), ('mute-for-session', N)], N), 'mute-pending')
check('domain state regression 1: muted', solve('muted', [('mute-window-ended', N, {'now': 29})], N, {'muted': True}, 'muted'), {'phase': 'muted', 'value': True})
check('domain state regression 2: muted', solve('muted', [('mute-window-ended', N, {'now': 30})], N, {'muted': True}, 'muted'), {'phase': 'unmute-pending', 'value': True})
check('domain state regression 3: extension', solve('muted', [('extend-mute', N, {'until': 20})], N, {}, 'extension'), {'phase': 'extend-pending', 'value': 30})
check('domain state regression 4: until', solve('extend-pending', [('extension-confirmed', N, {})], N, {}, 'until'), {'phase': 'muted-extended', 'value': 60})
check('domain state regression 5: muted', solve('muted-extended', [('old-window-ended', N, {})], N, {'until': 60, 'muted': True}, 'muted'), {'phase': 'muted-extended', 'value': True})
check('domain state regression 6: muted', solve('unmute-pending', [('unmute-confirmed', N, {})], N, {'muted': True}, 'muted'), {'phase': 'audible-channel', 'value': False})
check('domain state regression 7: until', solve('muted', [('extend-mute', N, {'until': 80}), ('extension-confirmed', N, {}), ('old-window-ended', N, {})], N, {'muted': True}, 'until'), {'phase': 'muted-extended', 'value': 80})
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 = {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'muted-extended', ('mute-error', 'cancel-mute-intent'): 'mute-pending'}
    data = json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})) if payload is None else {**json.loads(json.dumps({'until': 30, 'now': 10, 'muted': False, 'extension': 60})), **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 == 'mute-window-ended' and data['now'] < data['until'] and 'now' in arg: event = 'unknown-event'
            if event == 'extend-mute': data['extension'] = max(data['until'],arg.get('until',data['extension']))
            if event == 'extension-confirmed': data['until'] = data['extension']
            state = table.get((state, event), state)
            if state in ('muted','muted-extended'): data['muted'] = True
            if state == 'audible-channel': data['muted'] = False
    return {'phase': state, 'value': data[observe]} if observe else state
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('audible-channel / mute-for-session', solve('audible-channel', [('mute-for-session', N)], N), 'mute-pending')
check('mute-pending / mute-confirmed', solve('mute-pending', [('mute-confirmed', N)], N), 'muted')
check('mute-pending / mute-rejected', solve('mute-pending', [('mute-rejected', N)], N), 'mute-error')
check('muted / mute-window-ended', solve('muted', [('mute-window-ended', N)], N), 'unmute-pending')
check('unmute-pending / unmute-confirmed', solve('unmute-pending', [('unmute-confirmed', N)], N), 'audible-channel')
check('unmute-pending / unmute-rejected', solve('unmute-pending', [('unmute-rejected', N)], N), 'unmute-error')
check('muted / extend-mute', solve('muted', [('extend-mute', N)], N), 'extend-pending')
check('extend-pending / extension-confirmed', solve('extend-pending', [('extension-confirmed', N)], N), 'muted-extended')
check('muted-extended / old-window-ended', solve('muted-extended', [('old-window-ended', N)], N), 'muted-extended')
check('mute-error / cancel-mute-intent', solve('mute-error', [('cancel-mute-intent', N)], N), 'audible-channel')
check('older notification incarnation', solve('audible-channel', [('mute-for-session', N - 1)], N), 'audible-channel')
check('future notification incarnation', solve('audible-channel', [('mute-for-session', N + 1)], N), 'audible-channel')
check('unknown event is inert', solve('audible-channel', [('unknown-event', N)], N), 'audible-channel')
check('empty delivery batch', solve('audible-channel', [], N), 'audible-channel')
check('N stale deliveries before current delivery', solve('audible-channel', [('mute-for-session', N - 1)] * N + [('mute-for-session', N)], N), 'mute-pending')
check('trace mute-for-session then mute-confirmed', solve('audible-channel', [('mute-for-session', N), ('mute-confirmed', N)], N), 'muted')
check('trace mute-for-session then mute-rejected', solve('audible-channel', [('mute-for-session', N), ('mute-rejected', N)], N), 'mute-error')
check('trace mute-confirmed then mute-window-ended', solve('mute-pending', [('mute-confirmed', N), ('mute-window-ended', N)], N), 'unmute-pending')
check('trace mute-confirmed then extend-mute', solve('mute-pending', [('mute-confirmed', N), ('extend-mute', N)], N), 'extend-pending')
check('trace mute-rejected then cancel-mute-intent', solve('mute-pending', [('mute-rejected', N), ('cancel-mute-intent', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-confirmed', solve('muted', [('mute-window-ended', N), ('unmute-confirmed', N)], N), 'audible-channel')
check('trace mute-window-ended then unmute-rejected', solve('muted', [('mute-window-ended', N), ('unmute-rejected', N)], N), 'unmute-error')
check('trace unmute-confirmed then mute-for-session', solve('unmute-pending', [('unmute-confirmed', N), ('mute-for-session', N)], N), 'mute-pending')
check('trace extend-mute then extension-confirmed', solve('muted', [('extend-mute', N), ('extension-confirmed', N)], N), 'muted-extended')
check('trace extension-confirmed then old-window-ended', solve('extend-pending', [('extension-confirmed', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace old-window-ended then old-window-ended', solve('muted-extended', [('old-window-ended', N), ('old-window-ended', N)], N), 'muted-extended')
check('trace cancel-mute-intent then mute-for-session', solve('mute-error', [('cancel-mute-intent', N), ('mute-for-session', N)], N), 'mute-pending')
check('domain state regression 1: muted', solve('muted', [('mute-window-ended', N, {'now': 29})], N, {'muted': True}, 'muted'), {'phase': 'muted', 'value': True})
check('domain state regression 2: muted', solve('muted', [('mute-window-ended', N, {'now': 30})], N, {'muted': True}, 'muted'), {'phase': 'unmute-pending', 'value': True})
check('domain state regression 3: extension', solve('muted', [('extend-mute', N, {'until': 20})], N, {}, 'extension'), {'phase': 'extend-pending', 'value': 30})
check('domain state regression 4: until', solve('extend-pending', [('extension-confirmed', N, {})], N, {}, 'until'), {'phase': 'muted-extended', 'value': 60})
check('domain state regression 5: muted', solve('muted-extended', [('old-window-ended', N, {})], N, {'until': 60, 'muted': True}, 'muted'), {'phase': 'muted-extended', 'value': True})
check('domain state regression 6: muted', solve('unmute-pending', [('unmute-confirmed', N, {})], N, {'muted': True}, 'muted'), {'phase': 'audible-channel', 'value': False})
check('domain state regression 7: until', solve('muted', [('extend-mute', N, {'until': 80}), ('extension-confirmed', N, {}), ('old-window-ended', N, {})], N, {'muted': True}, 'until'), {'phase': 'muted-extended', 'value': 80})
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-mute-intent event leaves a notification in muted instead of audible-channel.

Temporary notification mute feedback is a bounded visual-notification workflow with mutable policy state {'until': 30, 'now': 10, 'muted': False, 'extension': 60}. Its default transition relation is {('audible-channel', 'mute-for-session'): 'mute-pending', ('mute-pending', 'mute-confirmed'): 'muted', ('mute-pending', 'mute-rejected'): 'mute-error', ('muted', 'mute-window-ended'): 'unmute-pending', ('unmute-pending', 'unmute-confirmed'): 'audible-channel', ('unmute-pending', 'unmute-rejected'): 'unmute-error', ('muted', 'extend-mute'): 'extend-pending', ('extend-pending', 'extension-confirmed'): 'muted-extended', ('muted-extended', 'old-window-ended'): 'muted-extended', ('mute-error', 'cancel-mute-intent'): 'audible-channel'}; 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 = {('valid', 'deadline-near'): 'expired', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 = {('valid', 'deadline-near'): 'valid', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 deadline-near event leaves a notification in expired instead of expiring.

Time-limited notification content expiry is a bounded visual-notification workflow with mutable policy state {'deadline': 20, 'now': 5, 'started': None, 'completed': False}. Its default transition relation is {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}; 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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'valid', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expiring', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 deadline-reached event leaves a notification in valid instead of expired.

Time-limited notification content expiry is a bounded visual-notification workflow with mutable policy state {'deadline': 20, 'now': 5, 'started': None, 'completed': False}. Its default transition relation is {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}; 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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'completed', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'expired', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 open-action event leaves a notification in completed instead of action-pending.

Time-limited notification content expiry is a bounded visual-notification workflow with mutable policy state {'deadline': 20, 'now': 5, 'started': None, 'completed': False}. Its default transition relation is {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}; 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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'expired', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 = {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'completed', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}
    data = json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})) if payload is None else {**json.loads(json.dumps({'deadline': 20, 'now': 5, 'started': None, 'completed': False})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            data['now'] = max(data['now'],arg.get('now',data['now']))
            if event == 'open-action':
                if data['now'] >= data['deadline']: event = 'unknown-event'
                else: data['started'] = data['now']
            if event == 'deadline-reached' and data['now'] < data['deadline'] and 'now' in arg: event = 'unknown-event'
            if event == 'extension-validated': data['deadline'] = max(data['deadline'],arg.get('deadline',data['deadline']+10))
            state = table.get((state, event), state)
            if state == 'completed': data['completed'] = True
    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('valid / deadline-near', solve('valid', [('deadline-near', N)], N), 'expiring')
check('expiring / deadline-reached', solve('expiring', [('deadline-reached', N)], N), 'expired')
check('valid / open-action', solve('valid', [('open-action', N)], N), 'action-pending')
check('action-pending / deadline-reached', solve('action-pending', [('deadline-reached', N)], N), 'pending-at-expiry')
check('pending-at-expiry / action-success', solve('pending-at-expiry', [('action-success', N)], N), 'completed')
check('pending-at-expiry / action-failure', solve('pending-at-expiry', [('action-failure', N)], N), 'expired-action-error')
check('expired / deadline-extension', solve('expired', [('deadline-extension', N)], N), 'revalidation-pending')
check('revalidation-pending / extension-validated', solve('revalidation-pending', [('extension-validated', N)], N), 'valid')
check('revalidation-pending / extension-rejected', solve('revalidation-pending', [('extension-rejected', N)], N), 'expired')
check('completed / late-expiry', solve('completed', [('late-expiry', N)], N), 'completed')
check('older notification incarnation', solve('valid', [('deadline-near', N - 1)], N), 'valid')
check('future notification incarnation', solve('valid', [('deadline-near', N + 1)], N), 'valid')
check('unknown event is inert', solve('valid', [('unknown-event', N)], N), 'valid')
check('empty delivery batch', solve('valid', [], N), 'valid')
check('N stale deliveries before current delivery', solve('valid', [('deadline-near', N - 1)] * N + [('deadline-near', N)], N), 'expiring')
check('trace deadline-near then deadline-reached', solve('valid', [('deadline-near', N), ('deadline-reached', N)], N), 'expired')
check('trace deadline-reached then deadline-extension', solve('expiring', [('deadline-reached', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace open-action then deadline-reached', solve('valid', [('open-action', N), ('deadline-reached', N)], N), 'pending-at-expiry')
check('trace deadline-reached then action-success', solve('action-pending', [('deadline-reached', N), ('action-success', N)], N), 'completed')
check('trace deadline-reached then action-failure', solve('action-pending', [('deadline-reached', N), ('action-failure', N)], N), 'expired-action-error')
check('trace action-success then late-expiry', solve('pending-at-expiry', [('action-success', N), ('late-expiry', N)], N), 'completed')
check('trace deadline-extension then extension-validated', solve('expired', [('deadline-extension', N), ('extension-validated', N)], N), 'valid')
check('trace deadline-extension then extension-rejected', solve('expired', [('deadline-extension', N), ('extension-rejected', N)], N), 'expired')
check('trace extension-validated then deadline-near', solve('revalidation-pending', [('extension-validated', N), ('deadline-near', N)], N), 'expiring')
check('trace extension-validated then open-action', solve('revalidation-pending', [('extension-validated', N), ('open-action', N)], N), 'action-pending')
check('trace extension-rejected then deadline-extension', solve('revalidation-pending', [('extension-rejected', N), ('deadline-extension', N)], N), 'revalidation-pending')
check('trace late-expiry then late-expiry', solve('completed', [('late-expiry', N), ('late-expiry', N)], N), 'completed')
check('domain state regression 1: started', solve('valid', [('open-action', N, {'now': 19})], N, {}, 'started'), {'phase': 'action-pending', 'value': 19})
check('domain state regression 2: started', solve('valid', [('open-action', N, {'now': 20})], N, {}, 'started'), {'phase': 'valid', 'value': None})
check('domain state regression 3: now', solve('expiring', [('deadline-reached', N, {'now': 19})], N, {}, 'now'), {'phase': 'expiring', 'value': 19})
check('domain state regression 4: completed', solve('pending-at-expiry', [('action-success', N, {})], N, {'started': 19, 'now': 21}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 5: deadline', solve('revalidation-pending', [('extension-validated', N, {'deadline': 15})], N, {}, 'deadline'), {'phase': 'valid', 'value': 20})
check('domain state regression 6: completed', solve('completed', [('late-expiry', N, {})], N, {'completed': True}, 'completed'), {'phase': 'completed', 'value': True})
check('domain state regression 7: started', solve('valid', [('open-action', N, {'now': 19}), ('deadline-reached', N, {'now': 20}), ('action-success', N, {})], N, {}, 'started'), {'phase': 'completed', 'value': 19})
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 deadline-reached event leaves a notification in expired instead of pending-at-expiry.

Time-limited notification content expiry is a bounded visual-notification workflow with mutable policy state {'deadline': 20, 'now': 5, 'started': None, 'completed': False}. Its default transition relation is {('valid', 'deadline-near'): 'expiring', ('expiring', 'deadline-reached'): 'expired', ('valid', 'open-action'): 'action-pending', ('action-pending', 'deadline-reached'): 'pending-at-expiry', ('pending-at-expiry', 'action-success'): 'completed', ('pending-at-expiry', 'action-failure'): 'expired-action-error', ('expired', 'deadline-extension'): 'revalidation-pending', ('revalidation-pending', 'extension-validated'): 'valid', ('revalidation-pending', 'extension-rejected'): 'expired', ('completed', 'late-expiry'): 'completed'}; 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