#!/usr/bin/env python3 """ Validate interaction for 'segfault' Typically run by verifyproblem. To run on its own from the command line, invoke as > ./validate.py input_file.in judge_answer.ans feedback_dir feedback_dir must exist, this is where the judgemessages are written judge_answer.ans is ignored """ from typing import Union from dataclasses import dataclass import random import re import sys from collections import defaultdict from pathlib import Path from typing import NoReturn def fail(msg: str, leak_to_team=False) -> NoReturn: """Fail WA with given message""" with open(sys.argv[3] / Path("judgemessage.txt"), "a", encoding="utf-8") as jfile: jfile.write(f"WA: {msg}\n") if leak_to_team: with open(sys.argv[3] / Path("teammessage.txt"), "a", encoding="utf-8") as afile: afile.write(f"{msg}\n") sys.exit(43) def accept() -> NoReturn: sys.exit(42) def get_team_line() -> str: team_bytes = sys.stdin.buffer.readline() try: team_line = team_bytes.decode("utf-8") except UnicodeDecodeError: fail("Team answer contains unexpected (non-utf-8) characters", leak_to_team=True) if not team_line: fail("Program exited before interaction was over", leak_to_team=True) return team_line def too_many_reads() -> NoReturn: print("Too many reads", flush=True) fail("Too many rounds") def segfault(index_s: str, answer: int = 10**18) -> NoReturn: print("Segmentation fault (core dumped)", flush=True) fail(f"Segmentation fault: {index_s} ∉ [0;{2*answer})") def wrong_answer(answer_given: str, expected_answer: int) -> NoReturn: fail(f"Got answer {answer_given}, expected {expected_answer}") @dataclass class QueryInput: index: int @dataclass class AnswerInput: answer: str def read_input() -> Union[QueryInput, AnswerInput]: line = get_team_line().strip() if match := RE_READ.fullmatch(line): index_s = match.group(1) # this prohibits "0" * 31 as a query for index 0 but w/e if len(index_s) > 30 or not (0 <= (index := int(index_s)) < 2 * 10**18): segfault(index_s) return QueryInput(index) elif match := RE_ANSWER.fullmatch(line): answer_s = match.group(1) return AnswerInput(answer_s) else: fail(f"Team query has invalid format: {line}", leak_to_team=True) def get_query_limit(answer: int) -> int: return 2 * (answer - 1).bit_length() RE_READ = re.compile(r"buf\s*\[\s*(-?[0-9]+)\s*\]", re.ASCII) RE_ANSWER = re.compile(r"strlen\(buf\)\s*=\s*(-?[0-9]+)", re.ASCII) def validate_constant(answer: int) -> None: query_limit = get_query_limit(answer) buffer = defaultdict(lambda: random.randint(1, 255)) queries = 0 while True: inp = read_input() if isinstance(inp, QueryInput): if queries == query_limit: too_many_reads() queries += 1 index = inp.index if index >= 2 * answer: segfault(str(index), answer) print(0 if index >= answer else buffer[index], flush=True) elif isinstance(inp, AnswerInput): if str(answer) != inp.answer: wrong_answer(inp.answer, answer) break def validate_adaptive() -> None: N_lo = 2 N_hi = 10**18 queries = 0 while True: queries_left = (get_query_limit(N_lo) - queries, get_query_limit(N_hi) - queries) print(f"{N_lo=}, {N_hi=}, {queries_left=}", file=sys.stderr) assert N_lo <= N_hi inp = read_input() if isinstance(inp, QueryInput): queries += 1 if queries > get_query_limit(N_lo): too_many_reads() index = inp.index if index < N_lo: # We must answer 1 print(1, flush=True) elif index >= 2 * N_lo: # Segfault as the answer could be N_lo segfault(str(inp.index), N_lo) elif index >= N_hi: # We must answer 0 print(0, flush=True) else: assert index < 2 * N_lo # Either 0 or 1 is a valid answer to this query # An optimal solution will know both N_lo and N_hi # and can thus start binary searching without fear of segfaults is_binary_search_possible = 2 * N_lo >= N_hi if is_binary_search_possible: # If we answer 0 the interval is limited to [N_lo, index] pos_0 = index - N_lo + 1 # If we answer 1 the interval is limited to [index + 1, N_hi] pos_1 = N_hi - (index + 1) + 1 # Choose the largest interval ans_0 = pos_0 >= pos_1 else: ans_0 = False if ans_0: print(0, flush=True) N_hi = index else: print(1, flush=True) N_lo = index + 1 elif isinstance(inp, AnswerInput): answers = {str(N_lo), str(N_hi)} wa_answers = answers - {inp.answer} if wa_answers: expected_answer = int(next(iter(wa_answers))) wrong_answer(inp.answer, expected_answer) break with open(sys.argv[1]) as in_file: contents = in_file.read().strip() if contents == "ADAPTIVE": validate_adaptive() else: answer = int(contents) validate_constant(answer) if not sys.stdin.buffer.readline(): accept() else: fail("Extra output")