diff --git a/doc/architecture/SPEC/providers/zcode.md b/doc/architecture/SPEC/providers/zcode.md index 89b9eac1..fc3cfd7f 100644 --- a/doc/architecture/SPEC/providers/zcode.md +++ b/doc/architecture/SPEC/providers/zcode.md @@ -1,8 +1,8 @@ # providers/zcode.py —— 规范 - + -`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` 存在，否则静默回退到目录里第一个模型——这就是过去手写条目"不起作用"的原因）， diff --git a/src/infermatrix_copilot/providers/zcode.py b/src/infermatrix_copilot/providers/zcode.py index 2c76255f..2eea6e26 100644 --- a/src/infermatrix_copilot/providers/zcode.py +++ b/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" diff --git a/test/test_zcode_stream_spool.py b/test/test_zcode_stream_spool.py new file mode 100644 index 00000000..b5061cf1 --- /dev/null +++ b/test/test_zcode_stream_spool.py @@ -0,0 +1,217 @@ +"""Offline producers exercise lossless native stdout spooling and cleanup.""" +from __future__ import annotations + +import json +import os +from pathlib import Path +import shutil +import stat +import sys +import tempfile +import time + +import pytest + +from infermatrix_copilot.providers import zcode +from infermatrix_copilot.trace_store import TraceStore + + +pytestmark = pytest.mark.skipif(sys.platform != "linux", reason="native process-group and /tmp checks are Linux-specific") + + +@pytest.fixture +def session(): + # The producer's spool must stay on local temporary storage, rather than + # inherit a network-backed working tree or test --basetemp. + with tempfile.TemporaryDirectory(prefix="imc-zcode-spool-test-", dir="/tmp") as raw: + path = Path(raw) + assert stat.S_IMODE(path.stat().st_mode) == 0o700 + yield path + + +@pytest.fixture +def spools(monkeypatch, session): + original = zcode.subprocess.Popen + observed = [] + + def checked_popen(*args, **kwargs): + if kwargs.get("cwd") != str(session) or not kwargs.get("start_new_session"): + return original(*args, **kwargs) + handles = [kwargs["stdout"], kwargs["stderr"]] + for handle in handles: + # A regular file makes Node's writes synchronous, including the + # producer's final process.exit(). It must not contain public data. + info = os.fstat(handle.fileno()) + assert stat.S_ISREG(info.st_mode) + assert stat.S_IMODE(info.st_mode) == 0o600 + target = os.readlink(f"/proc/self/fd/{handle.fileno()}") + assert target.startswith(str(session) + "/") + proc = original(*args, **kwargs) + observed.append((proc, handles)) + return proc + + monkeypatch.setattr(zcode.subprocess, "Popen", checked_popen) + return observed + + +def run_python(script, session, sink, *, timeout=5): + return zcode.ZCodeTransport._stream_run( + [sys.executable, "-c", script], session, os.environ.copy(), timeout, sink) + + +def assert_spools_closed(observed, session): + assert len(observed) == 1 + proc, handles = observed[0] + assert proc.poll() is not None + assert all(handle.closed for handle in handles) + assert list(session.iterdir()) == [] + + +@pytest.mark.skipif(shutil.which("node") is None, reason="Node is needed to reproduce forced-exit stdout loss") +def test_forced_node_exit_keeps_every_reasoning_delta_and_exact_result_with_slow_journal(session, spools, tmp_path): + count = 3000 + response = json.dumps({"operations": [], "rationale": "complete answer 完整"}, ensure_ascii=False) + script = "\n".join([ + "const count = 3000;", + "for (let i=0; i process.exit(0), 1000);", + ]) + store = TraceStore(tmp_path / "traces", environ={}) + archive = store.begin_call({"provider": "zcode", "model": "fixture", "system": "s", "prompt": "p"}) + + def slow_journal(event): + archive.event(event) + time.sleep(0.002) + + events, timed_out = zcode.ZCodeTransport._stream_run( + [shutil.which("node"), "-e", script], session, os.environ.copy(), 20, slow_journal) + + assert not timed_out + assert len(events) == count + 1 + assert [event["seq"] for event in events[:-1]] == list(range(count)) + assert events[-1] == {"type": "result", "response": response} + journal = [json.loads(line) for line in archive.events_path.read_text().splitlines()] + assert journal == events + assert_spools_closed(spools, session) + + +def test_completed_producer_is_not_timed_out_while_its_durable_sink_drains(session, spools): + count = 50 + script = "\n".join([ + "import json,os", + f"events=[{{'type':'model.streaming','seq':i,'payload':{{'kind':'reasoning_delta','delta':'partial'}}}} for i in range({count})]", + "events.append({'type':'result','response':'exact final answer'})", + "os.write(1, ('\\n'.join(json.dumps(e) for e in events)+'\\n').encode())", + ]) + received = [] + + def slow_sink(event): + received.append(event) + time.sleep(0.04) + + started = time.monotonic() + events, timed_out = run_python(script, session, slow_sink, timeout=1.0) + + assert time.monotonic() - started > 1.0 + assert not timed_out + assert events == received + assert len(events) == count + 1 + assert events[-1]["response"] == "exact final answer" + assert_spools_closed(spools, session) + + +def test_real_timeout_kills_owned_group_and_retains_received_partial_events(session, spools): + script = "\n".join([ + "import json,os,subprocess,sys,time", + "child=subprocess.Popen([sys.executable,'-c','import time; time.sleep(30)'])", + "events=[{'type':'session.updated','payload':{'child_pid':child.pid}}, {'type':'model.streaming','payload':{'kind':'reasoning_delta','delta':'retained partial'}}]", + "os.write(1, ('\\n'.join(json.dumps(e) for e in events)+'\\n').encode())", + "time.sleep(30)", + ]) + received = [] + events, timed_out = run_python(script, session, received.append, timeout=0.5) + + assert timed_out + assert events == received + assert events[-1]["payload"]["delta"] == "retained partial" + child_pid = events[0]["payload"]["child_pid"] + state = Path(f"/proc/{child_pid}/stat") + deadline = time.monotonic() + 2 + while state.exists() and state.read_text().split(") ", 1)[1].split()[0] != "Z" and time.monotonic() < deadline: + time.sleep(0.02) + assert not state.exists() or state.read_text().split(") ", 1)[1].split()[0] == "Z" + assert_spools_closed(spools, session) + + +def test_producer_deadline_still_kills_cli_while_event_sink_is_blocked(session, spools): + script = "\n".join([ + "import json,os,time", + "os.write(1, (json.dumps({'type':'model.streaming','payload':{'delta':'before blocked sink'}})+'\\n').encode())", + "time.sleep(30)", + ]) + received = [] + + def blocked_sink(event): + received.append(event) + time.sleep(0.9) + # The reader cannot check its deadline while this callback runs. The + # independent deadline guard must already have reaped the producer. + assert spools[0][0].poll() is not None + + events, timed_out = run_python(script, session, blocked_sink, timeout=0.3) + + assert timed_out + assert events == received + assert len(events) == 1 + assert events[0]["payload"]["delta"] == "before blocked sink" + assert_spools_closed(spools, session) + + +@pytest.mark.parametrize("returncode,with_result", [(3, True), (0, False)]) +def test_nonzero_or_missing_result_fails_closed_and_preserves_unterminated_tails(session, spools, returncode, with_result): + stdout_tail = '{"type":"model.streaming","payload":' + stderr_tail = "diagnostic without newline" + script = "\n".join([ + "import json,os", + "os.write(1,(json.dumps({'type':'model.streaming','payload':{'delta':'before tail'}})+'\\n').encode())", + *(["os.write(1,(json.dumps({'type':'result','response':'must reject nonzero exit'})+'\\n').encode())"] if with_result else []), + f"os.write(1,{stdout_tail.encode()!r})", + f"os.write(2,{stderr_tail.encode()!r})", + f"os._exit({returncode})", + ]) + received = [] + + with pytest.raises(RuntimeError, match=f"zcode exited {returncode} without a result event"): + run_python(script, session, received.append) + + assert {"type": "native.stdout", "text": stdout_tail} in received + assert {"type": "native.stderr", "text": stderr_tail} in received + assert received[0]["type"] == "model.streaming" + assert_spools_closed(spools, session) + + +def test_sink_interrupt_reaps_producer_drains_already_spooled_tails_and_closes_spools(session, spools): + tail = "received stderr tail" + script = "\n".join([ + "import json,os,time", + "events=[{'type':'model.streaming','seq':i,'payload':{'delta':'partial'}} for i in range(3)]", + f"os.write(2,{tail.encode()!r})", + "os.write(1, ('\\n'.join(json.dumps(e) for e in events)+'\\n').encode())", + "time.sleep(30)", + ]) + received = [] + interruption = KeyboardInterrupt("original sink interruption") + + def interrupted_sink(event): + received.append(event) + if len(received) == 1: + raise interruption + + with pytest.raises(KeyboardInterrupt) as caught: + run_python(script, session, interrupted_sink) + + assert caught.value is interruption + assert [event["seq"] for event in received if event["type"] == "model.streaming"] == [0, 1, 2] + assert {"type": "native.stderr", "text": tail} in received + assert_spools_closed(spools, session)