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

N = 1
observations = []
def solve(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.clear()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop(0)
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated announcement-transaction contract.

Nested announcement transactions stage text. Begin pushes a frame; append outside a frame speaks immediately. Commit merges into parent or speaks outer contents in order; rollback discards only the top frame. Replace edits only last staged item or appends when empty. Clear empties top frame. Flush cannot expose staged text. Report spoken and nested staged frames.

split: open-access


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

N = 1
observations = []
def solve(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][0]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1].append(e[1])
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated announcement-transaction contract.

Nested announcement transactions stage text. Begin pushes a frame; append outside a frame speaks immediately. Commit merges into parent or speaks outer contents in order; rollback discards only the top frame. Replace edits only last staged item or appends when empty. Clear empties top frame. Flush cannot expose staged text. Report spoken and nested staged frames.

split: open-access


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

N = 1
observations = []
def solve(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames.clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear(); spoken.clear()
        elif op=='flush': pass
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated announcement-transaction contract.

Nested announcement transactions stage text. Begin pushes a frame; append outside a frame speaks immediately. Commit merges into parent or speaks outer contents in order; rollback discards only the top frame. Replace edits only last staged item or appends when empty. Clear empties top frame. Flush cannot expose staged text. Report spoken and nested staged frames.

split: open-access


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

N = 1
observations = []
def solve(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': spoken.extend(x for f in frames for x in f)
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    frames=[]; spoken=[]
    for e in events:
        op=e[0]
        if op=='begin': frames.append([])
        elif op=='append':
            if frames: frames[-1].append(e[1])
            else: spoken.append(e[1])
        elif op=='commit' and frames:
            batch=frames.pop()
            if frames: frames[-1].extend(batch)
            else: spoken.extend(batch)
        elif op=='rollback' and frames: frames.pop()
        elif op=='replace' and frames:
            if frames[-1]: frames[-1][-1]=e[1]
            else: frames[-1].append(e[1])
        elif op=='clear' and frames: frames[-1].clear()
        elif op=='flush': spoken.extend(frames[-1] if frames else [])
    return [spoken,frames]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['append','outside'*N],['append','second'],['begin'],['append','first'],['append','second'],['replace','fixed'*N],['flush'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['commit'],['commit']]", "[['begin'],['append','outer'*N],['begin'],['append','inner'],['rollback'],['commit']]", "[['append','spoken'*N],['begin'],['append','hidden'],['clear'],['append','new'],['commit']]", "[['begin'],['replace','new'*N],['commit']]", "[['begin'],['append','x'*N],['clear']]", "[['commit'],['rollback'],['clear'],['flush']]", "[['begin'],['append','a'*N],['begin'],['append','b']]"]
expected = {1: [[['outside', 'second', 'first', 'fixed'], []], [['outer', 'inner'], []], [['outer'], []], [['spoken', 'new'], []], [['new'], []], [[], [[]]], [[], []], [[], [['a'], ['b']]]], 2: [[['outsideoutside', 'second', 'first', 'fixedfixed'], []], [['outerouter', 'inner'], []], [['outerouter'], []], [['spokenspoken', 'new'], []], [['newnew'], []], [[], [[]]], [[], []], [[], [['aa'], ['b']]]], 3: [[['outsideoutsideoutside', 'second', 'first', 'fixedfixedfixed'], []], [['outerouterouter', 'inner'], []], [['outerouterouter'], []], [['spokenspokenspoken', 'new'], []], [['newnewnew'], []], [[], [[]]], [[], []], [[], [['aaa'], ['b']]]], 4: [[['outsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixed'], []], [['outerouterouterouter', 'inner'], []], [['outerouterouterouter'], []], [['spokenspokenspokenspoken', 'new'], []], [['newnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaa'], ['b']]]], 5: [[['outsideoutsideoutsideoutsideoutside', 'second', 'first', 'fixedfixedfixedfixedfixed'], []], [['outerouterouterouterouter', 'inner'], []], [['outerouterouterouterouter'], []], [['spokenspokenspokenspokenspoken', 'new'], []], [['newnewnewnewnew'], []], [[], [[]]], [[], []], [[], [['aaaaa'], ['b']]]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated announcement-transaction contract.

Nested announcement transactions stage text. Begin pushes a frame; append outside a frame speaks immediately. Commit merges into parent or speaks outer contents in order; rollback discards only the top frame. Replace edits only last staged item or appends when empty. Clear empties top frame. Flush cannot expose staged text. Report spoken and nested staged frames.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+2
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],leases.get(e[1],[0,''])[1]]
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],e[1]]
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]>leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]<leases[e[1]][0]-1: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]+=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3][:1]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],e[1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1][:1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': pass
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.clear()
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=list(leases[e[1]])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1]); leases[e[2]][1]=""
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]<epoch: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]>epoch+1: continue
                if text: spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access


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

N = 1
observations = []
def solve(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                spoken.append(text)
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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(events):
    leases={}; epochs={}; queue=[]; spoken=[]
    for e in events:
        if e[0]=='register':
            epochs[e[1]]=epochs.get(e[1],0)+1
            leases[e[1]]=[epochs[e[1]],'']
        elif e[0]=='publish':
            if e[1] not in leases: continue
            if e[2]!=leases[e[1]][0]: continue
            leases[e[1]][1]=e[3]
        elif e[0]=='queue' and e[1] in leases:
            queue.append([e[1],leases[e[1]][0],leases[e[1]][1]])
        elif e[0]=='remove': leases.pop(e[1],None)
        elif e[0]=='rename' and e[1] in leases:
            leases[e[2]]=leases.pop(e[1])
        elif e[0]=='drain':
            for source,epoch,text in queue:
                if source not in leases: continue
                if leases[source][0]!=epoch: continue
                if text.strip(): spoken.append(text.strip())
            queue.clear()
    return [spoken,len(queue),sorted((k,v) for k,v in leases.items())]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
fixtures = ["[['register','a'],['publish','a',1,'old'*N],['queue','a'],['publish','a',1,'new'],['drain'],['drain']]", "[['register','a'],['publish','a',1,'old'],['queue','a'],['register','a'],['publish','a',1,'stale'*N],['queue','a'],['drain']]", "[['register','a'],['register','b'],['publish','b',1,'keep'*N],['remove','a'],['queue','b'],['drain']]", "[['register','a'],['publish','a',1,'hello'*N],['queue','a'],['rename','a','b'],['queue','b'],['drain']]", "[['register','a'],['queue','a'],['drain'],['publish','a',1,'  x  '*N],['queue','a'],['drain']]", "[['publish','missing',1,'x'*N],['remove','missing'],['drain']]", "[['register','a'],['publish','a',1,'x'*N],['queue','a'],['remove','a'],['drain']]"]
expected = {1: [[['old'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keep'], 0, [('b', [1, 'keep'])]], [['hello'], 0, [('b', [1, 'hello'])]], [['  x  '], 0, [('a', [1, '  x  '])]], [[], 0, []], [[], 0, []]], 2: [[['oldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeep'], 0, [('b', [1, 'keepkeep'])]], [['hellohello'], 0, [('b', [1, 'hellohello'])]], [['  x    x  '], 0, [('a', [1, '  x    x  '])]], [[], 0, []], [[], 0, []]], 3: [[['oldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeep'], 0, [('b', [1, 'keepkeepkeep'])]], [['hellohellohello'], 0, [('b', [1, 'hellohellohello'])]], [['  x    x    x  '], 0, [('a', [1, '  x    x    x  '])]], [[], 0, []], [[], 0, []]], 4: [[['oldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeep'])]], [['hellohellohellohello'], 0, [('b', [1, 'hellohellohellohello'])]], [['  x    x    x    x  '], 0, [('a', [1, '  x    x    x    x  '])]], [[], 0, []], [[], 0, []]], 5: [[['oldoldoldoldold'], 0, [('a', [1, 'new'])]], [[], 0, [('a', [2, ''])]], [['keepkeepkeepkeepkeep'], 0, [('b', [1, 'keepkeepkeepkeepkeep'])]], [['hellohellohellohellohello'], 0, [('b', [1, 'hellohellohellohellohello'])]], [['  x    x    x    x    x  '], 0, [('a', [1, '  x    x    x    x    x  '])]], [[], 0, []], [[], 0, []]]}[N]
for i, expression in enumerate(fixtures):
    check("trace-"+str(i+1), solve(eval(expression)), expected[i])
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 announcement trace violates the stated context-lease contract.

Announcement context leases map source to [epoch,text]. Register increments the source epoch and clears its text; publish requires a live source and matching epoch. Queue snapshots text plus source epoch. Drain speaks only still-live matching leases. Remove revokes source. Rename transfers a source lease and invalidates queued old-source messages. Report spoken and queue depth.

split: open-access