From 80dc2eb5c636eb9372881d23324b6bf2c3515f09 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 18 Oct 2026 07:58:28 +0000 Subject: [PATCH] scripts: index a gateway's log before its file is written (P1R-C06) The full recovery run on the qualification host failed on one termination sample whose release the harness never watched. The product released the attempt (`scope_terminated`); the harness had no attempt ID to watch. The gateway logged the launch when it started the target, but the harness's log reader wrote each line to the evidence before it parsed it. DG-1's cold load builds on the same volume as the evidence, and that interval's evidence writer lagged by up to 5.6 seconds. The reader was still waiting on the disk when the 5-second wait for the attempt ID ran out, so the sample went unwatched and the interval failed as unreleased. The reader now parses and indexes each line as the gateway writes it, then hands it to a thread of its own that writes the log file. A busy disk delays only the file. `Gateway.log()` returns what the gateway has logged whether or not the file has it yet, and the two scenarios that read a gateway's log use it. A log that cannot be written is reported to the bench's evidence writer as raw evidence that could not be written, so it makes a measurement interval inconclusive as the writer's own errors do. Ending or closing a gateway still waits until its log file is complete. The protocol states this in English and Korean. Co-Authored-By: Claude Opus 5.5 (1M context) --- docs/ko/recovery-qualification.md | 2 +- docs/recovery-qualification.md | 2 +- docs/translations.json | 4 +- scripts/qualify-devguard.py | 8 +-- scripts/qualify_host.py | 75 +++++++++++++++++++++----- scripts/tests/test_qualify_devguard.py | 58 ++++++++++++++++++++ 6 files changed, 129 insertions(+), 20 deletions(-) diff --git a/docs/ko/recovery-qualification.md b/docs/ko/recovery-qualification.md index 95d4a2a..34d5539 100644 --- a/docs/ko/recovery-qualification.md +++ b/docs/ko/recovery-qualification.md @@ -33,7 +33,7 @@ ## 하네스 실행 -하네스는 CS-RG 스위트처럼 qualification 호스트에서 실행합니다([하네스 실행](cs-rg-qualification.md#running)). 시스템 Python을 쓰고, 외부 패키지를 쓰지 않으며, 하네스 자신과 하네스가 시작하는 모든 프로세스는 최소 환경만 받습니다. 독립 Runner는 하네스가 운영자처럼 직접 시작합니다. `codespace-codex-runtime --lifecycle independent`를 자기 세션에서, 전용 소켓 디렉터리로 실행하고, 그 로그를 증거로 남깁니다. `required`이면 쓰는 authority의 DevGuard 설정을 주며, credential 파일은 경로만 넘기고 하네스는 읽지 않습니다. LaunchAgent로 설치하는 것은 없고 정책도 바꾸지 않습니다. gateway는 `--runner uds --runner-socket --runner-lifecycle independent`, 종료 의도, 작업 데이터베이스로 실행하며, `required`이면 `--devguard register`도 줍니다. +하네스는 CS-RG 스위트처럼 qualification 호스트에서 실행합니다([하네스 실행](cs-rg-qualification.md#running)). 시스템 Python을 쓰고, 외부 패키지를 쓰지 않으며, 하네스 자신과 하네스가 시작하는 모든 프로세스는 최소 환경만 받습니다. 독립 Runner는 하네스가 운영자처럼 직접 시작합니다. `codespace-codex-runtime --lifecycle independent`를 자기 세션에서, 전용 소켓 디렉터리로 실행하고, 그 로그를 증거로 남깁니다. `required`이면 쓰는 authority의 DevGuard 설정을 주며, credential 파일은 경로만 넘기고 하네스는 읽지 않습니다. LaunchAgent로 설치하는 것은 없고 정책도 바꾸지 않습니다. gateway는 `--runner uds --runner-socket --runner-lifecycle independent`, 종료 의도, 작업 데이터베이스로 실행하며, `required`이면 `--devguard register`도 줍니다. 관리 대상 실행의 attempt ID와 scope root는 gateway가 로그를 쓰는 대로 그 로그에서 가져오고, 그 로그는 별도 스레드에서 증거에 씁니다. 그래서 부하로 디스크가 바빠도 늦어지는 것은 로그 파일뿐이고 표본이 기다리는 attempt는 늦어지지 않으며, 쓰지 못한 로그는 쓰지 못한 원시 증거로 셉니다. | 스위트 | 보여 주는 것 | 추가 인자 | | --- | --- | --- | diff --git a/docs/recovery-qualification.md b/docs/recovery-qualification.md index c13760c..cf81671 100644 --- a/docs/recovery-qualification.md +++ b/docs/recovery-qualification.md @@ -33,7 +33,7 @@ The independent mode is opt-in. Making it the default is a separate decision, ou ## Running the harness -The harness runs on the qualification host as the CS-RG suites do ([running the harness](cs-rg-qualification.md#running)): the system Python, no third-party packages, and a minimal environment for itself and every process it starts. It starts each independent Runner itself, as an operator would: `codespace-codex-runtime --lifecycle independent` in its own session, with a private socket directory, its log kept as evidence, and for `required` the DevGuard settings of the authority it uses, the credential file given by its path, which the harness never reads. Nothing is installed as a LaunchAgent and no policy changes. Its gateways run with `--runner uds --runner-socket --runner-lifecycle independent`, an exit intent, an operations database and, for `required`, `--devguard register`. +The harness runs on the qualification host as the CS-RG suites do ([running the harness](cs-rg-qualification.md#running)): the system Python, no third-party packages, and a minimal environment for itself and every process it starts. It starts each independent Runner itself, as an operator would: `codespace-codex-runtime --lifecycle independent` in its own session, with a private socket directory, its log kept as evidence, and for `required` the DevGuard settings of the authority it uses, the credential file given by its path, which the harness never reads. Nothing is installed as a LaunchAgent and no policy changes. Its gateways run with `--runner uds --runner-socket --runner-lifecycle independent`, an exit intent, an operations database and, for `required`, `--devguard register`. It takes a governed launch's attempt ID and scope root from the gateway's log as the gateway writes it, and writes that log to the evidence on a thread of its own, so a disk the load keeps busy delays the log file, never the attempt a sample waits for; a log it cannot write counts as raw evidence that could not be written. | Suite | What it shows | Extra arguments | | --- | --- | --- | diff --git a/docs/translations.json b/docs/translations.json index 116fd71..861ccdc 100644 --- a/docs/translations.json +++ b/docs/translations.json @@ -249,8 +249,8 @@ "회귀", "후보-조합" ], - "source_sha256": "90775a4bde26e863cf717e0b0af77f9a492ca76cfc9a721e47b52dc97efcd6c9", - "translation_sha256": "537841d793983da1d7ac7bd3327355cae57e7b87e64d9deb9be3449a3ee8d1e9" + "source_sha256": "8adac9dd635d082db99988c2213538346a2944578f243b7b93addd025c7fc00e", + "translation_sha256": "40136e6153849dfa854f655621ac3c09ba8d73af35a9b6ada8afad4dcc5c3253" }, { "id": "overview", diff --git a/scripts/qualify-devguard.py b/scripts/qualify-devguard.py index 0f052e1..8be5d67 100644 --- a/scripts/qualify-devguard.py +++ b/scripts/qualify-devguard.py @@ -335,7 +335,7 @@ def gateway(self, name, mode='register', helper='release', socket=None, timeout_ transcript = self.evidence.file(prefix + '.mcp.jsonl') gateway = host.Gateway(self.args.mcp_bin, flags, host.minimal_env(self.work, env), self.evidence.file(prefix + '.gateway.log'), - record=lambda row: self.writer.write(transcript, row)) + record=lambda row: self.writer.write(transcript, row), failed=self.writer.failed) self.live.gateways.append(gateway) self.gateways.append(gateway) self.evidence.append(f'{self.prefix}/gateways.jsonl', {'name': name, 'pid': gateway.pid, 'argv': gateway.argv, @@ -2220,7 +2220,7 @@ def gateway(self, name, intent='detach', timeout_s=None, database=None, initiali transcript = self.evidence.file(prefix + '.mcp.jsonl') gateway = host.Gateway(self.args.mcp_bin, self.flags(intent), self.environment(timeout_s, database), self.evidence.file(prefix + '.gateway.log'), - record=lambda row: self.writer.write(transcript, row)) + record=lambda row: self.writer.write(transcript, row), failed=self.writer.failed) self.live.gateways.append(gateway) self.gateways.append(gateway) self.evidence.append(f'{self.prefix}/gateways.jsonl', { @@ -2841,7 +2841,7 @@ def recovery_writer_lock(bench): ended = bench.end(g1, 'eof') g3.initialize(timeout=60) claim = claimed_by(runner, g3.pid) - log = g3.log_path.read_text(errors='replace') + log = g3.log() checks += [check('a gateway started while the first one ended waited for it, then served', 'waiting for it to end' in log and 'the only writer of the operations database' in log and ended['returncode'] == 0, ended=ended), @@ -3240,7 +3240,7 @@ def start(index): checks += [check('exactly one of them serves', len(serving) == 1, outcomes=outcomes), check('the other refused to start, waiting for the writer', len(losers) == 1 and losers[0].process.returncode not in (0, None) - and 'has another writer' in losers[0].log_path.read_text(errors='replace'), + and 'has another writer' in losers[0].log(), returncode=[gateway.process.returncode for gateway in losers]), check('only the serving one claimed the Runner, under epoch 2', len(serving) == 1 and granted == [g1.pid, serving[0].pid], claims=runner.claims())] diff --git a/scripts/qualify_host.py b/scripts/qualify_host.py index 26b7b8b..a88089e 100644 --- a/scripts/qualify_host.py +++ b/scripts/qualify_host.py @@ -169,6 +169,11 @@ def __init__(self): def write(self, path, row): self.lines.put((time.monotonic(), Path(path), json.dumps(row) + '\n')) + def failed(self, message): + """Count raw evidence that another thread could not write, such as a gateway's log, as one + of this writer's errors.""" + self.errors.append(message) + def run(self): ending = False while not ending: @@ -479,18 +484,28 @@ def runner_flags(mode, runtime_bin=None, runner_dir=None, runner_socket=None): class Gateway: """A codespace-mcp process on stdio in its own session. Its log is kept as evidence, and the governed executions it reports are indexed by process ID, every report kept, so that a - second launch of one execution shows. Every process it starts through `exec` or a resumed - approval is tracked, so `settle` can end them before the gateway goes.""" - - def __init__(self, binary, flags, env, log_path, record=None, cwd=None): + second launch of one execution shows. The log's reader indexes each line as the gateway + writes it and hands it to a thread of its own that writes the log file, as the Writer keeps + samples off the disk: a disk that the load keeps busy delays the file, never the index the + harness waits on. A log that cannot be written is reported to `failed`, as raw evidence that + could not be written. Every process it starts through `exec` or a resumed approval is + tracked, so `settle` can end them before the gateway goes.""" + + def __init__(self, binary, flags, env, log_path, record=None, cwd=None, failed=None): self.argv = [str(binary), *flags] self.log_path = Path(log_path) + self.failed = failed self.executions = {} + self.lines = [] + self.log_error = None self.started = set() self.lock = threading.Lock() self.process = subprocess.Popen(self.argv, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, cwd=cwd, start_new_session=True) self.pid = self.process.pid + self.unwritten = queue.Queue() + self.log_writer = threading.Thread(target=self._write_log, daemon=True) + self.log_writer.start() self.log_reader = threading.Thread(target=self._log, daemon=True) self.log_reader.start() self.client = McpClient(self.process, record) @@ -498,21 +513,57 @@ def __init__(self, binary, flags, env, log_path, record=None, cwd=None): def _log(self): try: - with open(self.log_path, 'ab') as log: - for line in self.process.stderr: - log.write(line) - log.flush() - entry = launch_line(line.decode('utf-8', 'replace')) + for line in self.process.stderr: + text = line.decode('utf-8', 'replace') + entry = launch_line(text) + with self.lock: + self.lines.append(text) if entry: - with self.lock: - self.executions.setdefault(entry['process_id'], []).append(entry) + self.executions.setdefault(entry['process_id'], []).append(entry) + self.unwritten.put(line) finally: + self.unwritten.put(None) close_quietly(self.process.stderr) + def _write_log(self): + """Append the lines the reader took to the log file as fast as the disk takes them, and + close it once the reader has read to the end. A log that cannot be written keeps its + error and reports it; the reader goes on.""" + ended = False + try: + with open(self.log_path, 'ab') as log: + while not ended: + lines = [self.unwritten.get()] + while lines[-1] is not None: + try: + lines.append(self.unwritten.get_nowait()) + except queue.Empty: + break + ended = lines[-1] is None + log.write(b''.join(lines[:-1] if ended else lines)) + log.flush() + except OSError as error: + self.log_error = f'{self.log_path.name}: {error}' + if self.failed is not None: + self.failed(self.log_error) + while not ended: + ended = self.unwritten.get() is None + + def log(self, limit=10.0): + """What the gateway has logged so far, as its reader took it, whether or not the log file + has it yet: all of it once the gateway has exited, waiting up to `limit` seconds for the + reader to reach the end.""" + if self.process.poll() is not None: + self.log_reader.join(timeout=limit) + with self.lock: + return ''.join(self.lines) + def join_readers(self, limit=10): - """Wait for the readers of the gateway's output and log, which close its pipes as they end.""" + """Wait for the readers of the gateway's output and log, which close its pipes as they end, + and for the log's writer, which closes the log file once it has written every line.""" self.log_reader.join(timeout=limit) self.client.reader.join(timeout=limit) + self.log_writer.join(timeout=limit) def initialize(self, timeout=60): result, latency = self.client.request('initialize', { diff --git a/scripts/tests/test_qualify_devguard.py b/scripts/tests/test_qualify_devguard.py index d18906c..bf8a15f 100644 --- a/scripts/tests/test_qualify_devguard.py +++ b/scripts/tests/test_qualify_devguard.py @@ -407,6 +407,10 @@ def text(value, error=False): send({'jsonrpc': '2.0', 'id': 'server-1', 'method': 'ping'}) reply = json.loads(sys.stdin.readline()) send({'jsonrpc': '2.0', 'id': message['id'], 'result': text({'pinged': reply.get('result') == {}})}) + elif name == 'log': + sys.stderr.write(message['params']['arguments']['line'] + '\\n') + sys.stderr.flush() + send({'jsonrpc': '2.0', 'id': message['id'], 'result': text({})}) elif name == 'silent': pass elif name == 'exit': @@ -452,6 +456,60 @@ def test_no_answer_and_a_lost_connection(self): self.assertTrue(raised.exception.transport) +class GatewayLogTests(unittest.TestCase): + """A gateway's log is indexed as the gateway writes it, whatever the disk under its file does: + the harness waits on the index for a governed launch's attempt, and under DG-1's load the + evidence disk stalls for seconds.""" + + def setUp(self): + self.scratch = tempfile.TemporaryDirectory() + self.dir = Path(self.scratch.name) + self.server = self.dir / 'server.py' + self.server.write_text(FAKE_SERVER) + + def tearDown(self): + self.scratch.cleanup() + + def test_a_stalled_log_file_never_delays_the_index(self): + # Opening a FIFO to write waits until it has a reader, as a file on a stalled disk does. + path = self.dir / 'stalled.log' + os.mkfifo(path) + gateway = host.Gateway(sys.executable, [str(self.server)], dict(os.environ), path) + try: + gateway.initialize(timeout=30) + gateway.call('log', {'line': LAUNCHED}) + self.assertEqual(gateway.execution('proc-7c1d', limit=5), + {'attempt_id': 'cs-attempt-1f2e', 'process_id': 'proc-7c1d', 'launched': True, + 'scope_root_pid': 4242, 'dispatch': 'Confirmed'}) + self.assertEqual(gateway.log(), LAUNCHED + '\n') + self.assertTrue(gateway.log_writer.is_alive()) + finally: + reader = os.open(path, os.O_RDONLY | os.O_NONBLOCK) + gateway.close(limit=10) + written = b'' + try: + while chunk := os.read(reader, 65536): + written += chunk + finally: + os.close(reader) + self.assertEqual(written, (LAUNCHED + '\n').encode()) + self.assertFalse(gateway.log_writer.is_alive()) + self.assertIsNone(gateway.log_error) + + def test_a_log_that_cannot_be_written_counts_as_unwritten_evidence(self): + writer = host.Writer() + gateway = host.Gateway(sys.executable, [str(self.server)], dict(os.environ), self.dir / 'gone' / 'gw.log', + failed=writer.failed) + gateway.initialize(timeout=30) + gateway.call('log', {'line': LAUNCHED}) + self.assertEqual(gateway.execution('proc-7c1d', limit=5)['attempt_id'], 'cs-attempt-1f2e') + gateway.close(limit=10) + self.assertEqual(gateway.log(), LAUNCHED + '\n') + self.assertTrue(gateway.log_error.startswith('gw.log: ')) + report = writer.close() + self.assertEqual((report['error_count'], report['errors']), (1, [gateway.log_error])) + + FAKE_FIXTURE = textwrap.dedent(""" import json, sys print(json.dumps({'base': sys.argv[1]}), flush=True)