scripts: index a gateway's log before its file is written (P1R-C06)

docs/ko/recovery-qualification.md

@@ -33,7 +33,7 @@
 
 ## 하네스 실행
 
-하네스는 CS-RG 스위트처럼 qualification 호스트에서 실행합니다([하네스 실행](cs-rg-qualification.md#running)). 시스템 Python을 쓰고, 외부 패키지를 쓰지 않으며, 하네스 자신과 하네스가 시작하는 모든 프로세스는 최소 환경만 받습니다. 독립 Runner는 하네스가 운영자처럼 직접 시작합니다. `codespace-codex-runtime <socket> --lifecycle independent`를 자기 세션에서, 전용 소켓 디렉터리로 실행하고, 그 로그를 증거로 남깁니다. `required`이면 쓰는 authority의 DevGuard 설정을 주며, credential 파일은 경로만 넘기고 하네스는 읽지 않습니다. LaunchAgent로 설치하는 것은 없고 정책도 바꾸지 않습니다. gateway는 `--runner uds --runner-socket <socket> --runner-lifecycle independent`, 종료 의도, 작업 데이터베이스로 실행하며, `required`이면 `--devguard register`도 줍니다.
+하네스는 CS-RG 스위트처럼 qualification 호스트에서 실행합니다([하네스 실행](cs-rg-qualification.md#running)). 시스템 Python을 쓰고, 외부 패키지를 쓰지 않으며, 하네스 자신과 하네스가 시작하는 모든 프로세스는 최소 환경만 받습니다. 독립 Runner는 하네스가 운영자처럼 직접 시작합니다. `codespace-codex-runtime <socket> --lifecycle independent`를 자기 세션에서, 전용 소켓 디렉터리로 실행하고, 그 로그를 증거로 남깁니다. `required`이면 쓰는 authority의 DevGuard 설정을 주며, credential 파일은 경로만 넘기고 하네스는 읽지 않습니다. LaunchAgent로 설치하는 것은 없고 정책도 바꾸지 않습니다. gateway는 `--runner uds --runner-socket <socket> --runner-lifecycle independent`, 종료 의도, 작업 데이터베이스로 실행하며, `required`이면 `--devguard register`도 줍니다. 관리 대상 실행의 attempt ID와 scope root는 gateway가 로그를 쓰는 대로 그 로그에서 가져오고, 그 로그는 별도 스레드에서 증거에 씁니다. 그래서 부하로 디스크가 바빠도 늦어지는 것은 로그 파일뿐이고 표본이 기다리는 attempt는 늦어지지 않으며, 쓰지 못한 로그는 쓰지 못한 원시 증거로 셉니다.
 
 | 스위트 | 보여 주는 것 | 추가 인자 |
 | --- | --- | --- |

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 <socket> --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 <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 <socket> --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 <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 |
 | --- | --- | --- |

docs/translations.json

@@ -249,8 +249,8 @@
         "회귀",
         "후보-조합"
       ],
-      "source_sha256": "90775a4bde26e863cf717e0b0af77f9a492ca76cfc9a721e47b52dc97efcd6c9",
-      "translation_sha256": "537841d793983da1d7ac7bd3327355cae57e7b87e64d9deb9be3449a3ee8d1e9"
+      "source_sha256": "8adac9dd635d082db99988c2213538346a2944578f243b7b93addd025c7fc00e",
+      "translation_sha256": "40136e6153849dfa854f655621ac3c09ba8d73af35a9b6ada8afad4dcc5c3253"
     },
     {
       "id": "overview",

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())]

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,40 +484,86 @@ 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)
         self.info = 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', {

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)