fix(zcode): preserve completed results through slow native journals

doc/architecture/SPEC/providers/zcode.md

@@ -1,8 +1,8 @@
 # providers/zcode.py —— 规范
 
-<!-- verified-against: 2026-10-04 -->
+<!-- verified-against: 2026-10-09 -->
 
-`LOC ~400 · harness transport（Z.AI / GLM 订阅） · refactor-status: ok`
+`harness transport（Z.AI / GLM 订阅） · refactor-status: ok`
 
 ## 职责
 在 Z.AI OAuth 订阅认证下，通过 `zcode` CLI 跑完**一整个** agent step。
@@ -34,6 +34,11 @@ argv 单参数上限 128KiB）。
   放入会话临时目录；避免保留下来的 Read 接触宿主自动记忆。工具桥会话同样关闭 memory/use。
 - **失败的运行不能读成空评审。** 非零退出或流里没有 `result` 事件（未超时时）直接抛
   `RuntimeError`，带上 zcode 自己的 stderr。
+- **原生日志的持久化不能截断 CLI 输出。** 原生事件模式将 stdout/stderr 接到每次调用自有的
+  临时普通文件，宿主按独立偏移实时读取并逐事件留痕；不对共享写描述符 seek。
+  ZCode 的退出 watchdog 不等待管道缓冲排空，慢速留痕时普通文件避免丢失已完成回答。
+  进程结束后仍排空所有输出及末尾非换行片段，再严格核验 `result`；超时只约束仍在运行的
+  CLI，已退出进程的日志排空不误判为模型超时。异常／中断仍回收该调用的进程组和临时文件。
 - **模型按次钉死，再断言。** zcode 没有 `--model`；一个会话取其个人 provider 配置里的
   `defaultModelSelection`，且仅当该条目**可选**（0.16.9 的 `resolveInitialModelSelection` 要求
   `options.reasoningLevel` 存在，否则静默回退到目录里第一个模型——这就是过去手写条目"不起作用"的原因），

src/infermatrix_copilot/providers/zcode.py

@@ -56,13 +56,13 @@
 import logging
 import os
 import re
-import selectors
 import signal
 import time
 import shutil
 import subprocess
 import sys
 import tempfile
+import threading
 from functools import cached_property
 from pathlib import Path
 
@@ -321,12 +321,15 @@ def _run(self, text: str, *, session: Path, timeout_s: float,
 
     @staticmethod
     def _stream_run(cmd, session, env, timeout_s, sink):
-        """Drain both native pipes while journaling, including failed runs."""
-        proc = subprocess.Popen(cmd, cwd=str(session), env=env, stdin=subprocess.DEVNULL,
-                                stdout=subprocess.PIPE, stderr=subprocess.PIPE, start_new_session=True)
+        """Journal private regular-file spools without backpressuring the CLI.
+
+        ZCode writes stream events without awaiting pipe backpressure, then
+        forces process exit after a short grace period. A durable event sink
+        can therefore lose even a completed result through an output pipe.
+        POSIX regular-file writes are synchronous in Node; anonymous files
+        keep the full output while we journal it at the sink's own pace.
+        """
         events, stderr, buffers = [], [], {"stdout": b"", "stderr": b""}
-        deadline = time.monotonic() + timeout_s
-        timed_out = False
 
         def line(channel, raw):
             text = raw.decode("utf-8", "replace")
@@ -343,54 +346,102 @@ def line(channel, raw):
                 stderr.append(text)
             sink({"type": "native." + channel, "text": text})
 
-        def kill():
-            try:
-                os.killpg(proc.pid, signal.SIGKILL)
-            except ProcessLookupError:
-                pass
-
-        try:
-            with selectors.DefaultSelector() as selector:
-                selector.register(proc.stdout, selectors.EVENT_READ, "stdout")
-                selector.register(proc.stderr, selectors.EVENT_READ, "stderr")
-                while selector.get_map():
-                    remaining = deadline - time.monotonic()
-                    if remaining <= 0 and not timed_out:
+        # TemporaryFile creates mode-0600 anonymous files in our session;
+        # no named raw-output file persists or enters session attachments.
+        with tempfile.TemporaryFile(mode="w+b", buffering=0, dir=session) as stdout, \
+                tempfile.TemporaryFile(mode="w+b", buffering=0, dir=session) as stderr_file:
+            spools = {"stdout": stdout, "stderr": stderr_file}
+            offsets = {channel: 0 for channel in spools}
+            proc = subprocess.Popen(cmd, cwd=str(session), env=env, stdin=subprocess.DEVNULL,
+                                    stdout=stdout, stderr=stderr_file, start_new_session=True)
+            timed_out = False
+            group_stopped = False
+            stop_lock = threading.Lock()
+
+            def kill(*, at_deadline=False):
+                nonlocal group_stopped, timed_out
+                with stop_lock:
+                    if group_stopped or (at_deadline and proc.poll() is not None):
+                        return
+                    group_stopped = True
+                    if at_deadline:
                         timed_out = True
+                    try:
+                        os.killpg(proc.pid, signal.SIGKILL)
+                    except ProcessLookupError:
+                        pass
+
+            # A slow sink must neither postpone a real producer timeout nor
+            # turn draining an already-exited producer into a false timeout.
+            watchdog = threading.Timer(timeout_s, lambda: kill(at_deadline=True))
+            watchdog.daemon = True
+            watchdog_started = False
+
+            def emit(channel, raw, *, best_effort=False):
+                if not best_effort:
+                    line(channel, raw)
+                    return
+                try:
+                    line(channel, raw)
+                except BaseException:
+                    pass  # retain the original interruption / sink failure
+
+            def drain(channel, *, best_effort=False):
+                # pread must not move the shared child descriptor's offset:
+                # seek/read on it could make the CLI overwrite prior output.
+                chunk = os.pread(spools[channel].fileno(), 65536, offsets[channel])
+                offsets[channel] += len(chunk)
+                buffers[channel] += chunk
+                while b"\n" in buffers[channel]:
+                    raw, buffers[channel] = buffers[channel].split(b"\n", 1)
+                    emit(channel, raw, best_effort=best_effort)
+                return bool(chunk)
+
+            def tails(*, best_effort=False):
+                for channel in buffers:
+                    raw, buffers[channel] = buffers[channel], b""
+                    if raw:
+                        emit(channel, raw, best_effort=best_effort)
+
+            try:
+                watchdog.start()
+                watchdog_started = True
+                while True:
+                    alive = proc.poll() is None
+                    if not alive:
+                        # The leader has finished; no owned descendant may
+                        # keep growing a spool after its completed invocation.
                         kill()
-                    for key, _ in selector.select(0.25 if timed_out else min(1, max(0, remaining))):
-                        chunk = os.read(key.fileobj.fileno(), 65536)
-                        channel = key.data
-                        if not chunk:
-                            if buffers[channel]:
-                                line(channel, buffers[channel])
-                            buffers[channel] = b""
-                            selector.unregister(key.fileobj)
-                            continue
-                        buffers[channel] += chunk
-                        while b"\n" in buffers[channel]:
-                            raw, buffers[channel] = buffers[channel].split(b"\n", 1)
-                            line(channel, raw)
-            proc.wait()
-        except BaseException:
-            # A stream interruption can arrive between a pipe read and its
-            # next newline. Preserve every tail we already received, while
-            # retaining the original exception if the journal itself failed.
-            for channel, raw in buffers.items():
-                for tail in raw.splitlines():
+                    changed = False
+                    for channel in spools:
+                        changed = drain(channel) or changed
+                    if not alive and not changed:
+                        tails()
+                        break
+                    if not changed:
+                        time.sleep(0.02)
+                proc.wait()
+            except BaseException:
+                # Stop production first, then preserve all bytes already
+                # spooled, including tails lacking a final newline. Sink
+                # failures must never replace the original interruption.
+                kill()
+                proc.wait()
+                for channel in spools:
                     try:
-                        line(channel, tail)
+                        while drain(channel, best_effort=True):
+                            pass
                     except BaseException:
                         pass
-                buffers[channel] = b""
-            raise
-        finally:
-            # This group belongs only to this invocation, including children
-            # left after its CLI leader exits or our event writer is interrupted.
-            kill()
-            proc.wait()
-            proc.stdout.close()
-            proc.stderr.close()
+                tails(best_effort=True)
+                raise
+            finally:
+                # This process group belongs only to this invocation.
+                watchdog.cancel()
+                if watchdog_started:
+                    watchdog.join()
+                kill()
+                proc.wait()
         if not timed_out and (proc.returncode != 0 or not any(e.get("type") == "result" for e in events)):
             detail = " ".join(stderr[-3:])[:400]
             raise RuntimeError(f"zcode exited {proc.returncode} without a result event"