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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'warning-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'resolved', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 ack-critical event leaves a notification in warning-acknowledged instead of critical-acknowledged.

Notification severity escalation acknowledgement is a bounded visual-notification workflow with mutable policy state {'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1}. Its default transition relation is {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}; 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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'warning-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'informational', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 warning-arrives event leaves a notification in warning-acknowledged instead of critical-acknowledged.

Notification severity escalation acknowledgement is a bounded visual-notification workflow with mutable policy state {'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1}. Its default transition relation is {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}; 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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'warning-unacknowledged', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 condition-cleared event leaves a notification in resolved instead of resolved-unread.

Notification severity escalation acknowledgement is a bounded visual-notification workflow with mutable policy state {'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1}. Its default transition relation is {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}; 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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'critical-acknowledged', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'informational', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 condition-cleared event leaves a notification in critical-acknowledged instead of resolved.

Notification severity escalation acknowledgement is a bounded visual-notification workflow with mutable policy state {'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1}. Its default transition relation is {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}; 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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'critical-acknowledged', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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 = {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'informational', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}
    data = json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})) if payload is None else {**json.loads(json.dumps({'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1})), **json.loads(json.dumps(payload))}
    state = initial
    for delivery in events:
        event, event_generation = delivery[:2]
        arg = delivery[2] if len(delivery) > 2 else {}
        if event_generation == generation:
            previous = state
            if event in ('ack-warning','ack-critical') and arg.get('occurrence',data['occurrence']) != data['occurrence']: event = 'unknown-event'
            if event == 'recurrence': data['occurrence'] += 1
            state = table.get((state, event), state)
            if state.startswith('critical-'): data['level'] = 3
            if state.startswith('warning-'): data['level'] = 2
            if state.endswith('-acknowledged') and data['occurrence'] not in data['acknowledged']: data['acknowledged'].append(data['occurrence'])
            if state.startswith('resolved'): data['active'] = False
            if state == 'warning-unacknowledged': data['active'] = 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('informational / warning-arrives', solve('informational', [('warning-arrives', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / ack-warning', solve('warning-unacknowledged', [('ack-warning', N)], N), 'warning-acknowledged')
check('warning-acknowledged / critical-arrives', solve('warning-acknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('critical-unacknowledged / ack-critical', solve('critical-unacknowledged', [('ack-critical', N)], N), 'critical-acknowledged')
check('critical-acknowledged / warning-arrives', solve('critical-acknowledged', [('warning-arrives', N)], N), 'critical-acknowledged')
check('critical-unacknowledged / condition-cleared', solve('critical-unacknowledged', [('condition-cleared', N)], N), 'resolved-unread')
check('critical-acknowledged / condition-cleared', solve('critical-acknowledged', [('condition-cleared', N)], N), 'resolved')
check('resolved-unread / open-resolution', solve('resolved-unread', [('open-resolution', N)], N), 'resolved')
check('resolved / recurrence', solve('resolved', [('recurrence', N)], N), 'warning-unacknowledged')
check('warning-unacknowledged / critical-arrives', solve('warning-unacknowledged', [('critical-arrives', N)], N), 'critical-unacknowledged')
check('older notification incarnation', solve('informational', [('warning-arrives', N - 1)], N), 'informational')
check('future notification incarnation', solve('informational', [('warning-arrives', N + 1)], N), 'informational')
check('unknown event is inert', solve('informational', [('unknown-event', N)], N), 'informational')
check('empty delivery batch', solve('informational', [], N), 'informational')
check('N stale deliveries before current delivery', solve('informational', [('warning-arrives', N - 1)] * N + [('warning-arrives', N)], N), 'warning-unacknowledged')
check('trace warning-arrives then ack-warning', solve('informational', [('warning-arrives', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace warning-arrives then critical-arrives', solve('informational', [('warning-arrives', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace ack-warning then critical-arrives', solve('warning-unacknowledged', [('ack-warning', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-acknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-acknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('trace ack-critical then warning-arrives', solve('critical-unacknowledged', [('ack-critical', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace ack-critical then condition-cleared', solve('critical-unacknowledged', [('ack-critical', N), ('condition-cleared', N)], N), 'resolved')
check('trace warning-arrives then warning-arrives', solve('critical-acknowledged', [('warning-arrives', N), ('warning-arrives', N)], N), 'critical-acknowledged')
check('trace warning-arrives then condition-cleared', solve('critical-acknowledged', [('warning-arrives', N), ('condition-cleared', N)], N), 'resolved')
check('trace condition-cleared then open-resolution', solve('critical-unacknowledged', [('condition-cleared', N), ('open-resolution', N)], N), 'resolved')
check('trace condition-cleared then recurrence', solve('critical-acknowledged', [('condition-cleared', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace open-resolution then recurrence', solve('resolved-unread', [('open-resolution', N), ('recurrence', N)], N), 'warning-unacknowledged')
check('trace recurrence then ack-warning', solve('resolved', [('recurrence', N), ('ack-warning', N)], N), 'warning-acknowledged')
check('trace recurrence then critical-arrives', solve('resolved', [('recurrence', N), ('critical-arrives', N)], N), 'critical-unacknowledged')
check('trace critical-arrives then ack-critical', solve('warning-unacknowledged', [('critical-arrives', N), ('ack-critical', N)], N), 'critical-acknowledged')
check('trace critical-arrives then condition-cleared', solve('warning-unacknowledged', [('critical-arrives', N), ('condition-cleared', N)], N), 'resolved-unread')
check('domain state regression 1: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 0})], N, {}, 'acknowledged'), {'phase': 'critical-unacknowledged', 'value': []})
check('domain state regression 2: acknowledged', solve('critical-unacknowledged', [('ack-critical', N, {'occurrence': 1})], N, {}, 'acknowledged'), {'phase': 'critical-acknowledged', 'value': [1]})
check('domain state regression 3: level', solve('warning-acknowledged', [('critical-arrives', N, {})], N, {'level': 2}, 'level'), {'phase': 'critical-unacknowledged', 'value': 3})
check('domain state regression 4: level', solve('critical-acknowledged', [('warning-arrives', N, {})], N, {'level': 3}, 'level'), {'phase': 'critical-acknowledged', 'value': 3})
check('domain state regression 5: active', solve('critical-unacknowledged', [('condition-cleared', N, {})], N, {}, 'active'), {'phase': 'resolved-unread', 'value': False})
check('domain state regression 6: occurrence', solve('resolved', [('recurrence', N, {})], N, {'acknowledged': [1], 'active': False}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
check('domain state regression 7: occurrence', solve('warning-unacknowledged', [('ack-warning', N, {}), ('critical-arrives', N, {}), ('condition-cleared', N, {}), ('open-resolution', N, {}), ('recurrence', N, {})], N, {}, 'occurrence'), {'phase': 'warning-unacknowledged', 'value': 2})
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-resolution event leaves a notification in critical-acknowledged instead of resolved.

Notification severity escalation acknowledgement is a bounded visual-notification workflow with mutable policy state {'occurrence': 1, 'acknowledged': [], 'active': True, 'level': 1}. Its default transition relation is {('informational', 'warning-arrives'): 'warning-unacknowledged', ('warning-unacknowledged', 'ack-warning'): 'warning-acknowledged', ('warning-acknowledged', 'critical-arrives'): 'critical-unacknowledged', ('critical-unacknowledged', 'ack-critical'): 'critical-acknowledged', ('critical-acknowledged', 'warning-arrives'): 'critical-acknowledged', ('critical-unacknowledged', 'condition-cleared'): 'resolved-unread', ('critical-acknowledged', 'condition-cleared'): 'resolved', ('resolved-unread', 'open-resolution'): 'resolved', ('resolved', 'recurrence'): 'warning-unacknowledged', ('warning-unacknowledged', 'critical-arrives'): 'critical-unacknowledged'}; 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