From 0cd77d6fb0641020c55e2b15b7cc3586e7a8db74 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Wed, 28 Oct 2026 11:44:25 +0800 Subject: [PATCH] Graceful SIGINT/SIGTERM shutdown, resilient tool bridge, wrapper signal receipts Three defects from run-20261027-193251 (vllm-omni rebase campaign): 1. imx-omni-pytest wrappers reported child rc -15/-9/-2 with neither timeout nor watchdog set: the host GPU reservation guard killed the test processes after the campaign's reservation was lost, and the receipt could not say so. The wrapper now runs in its own session, ignores SIGHUP, stops its tree on SIGTERM/SIGINT, and records signal / signal_origin (timeout, watchdog, wrapper_signal, external) plus pid/pgid/sid/ppid context in a schema-2 receipt and a log footer; tests spawn with stdin=/dev/null; the live prompt's detach + poll procedure is explicit and verifiable. 2. infermatrix-tools MCP discovery failed mid-attempt: FastMCP ran the sync handlers inline, so a 30-minute plan review froze the server and the client dropped it. Handlers are offloaded to worker threads, the bridge ignores SIGINT/SIGHUP/SIGPIPE, logs start/slow_tool/exit(reason) to bridge/.lifecycle.jsonl, and the module loop records a bridge_unavailable trace event with the bridge's exit reason. 3. The engine inherited SIG_IGN for SIGINT from its nohup launcher and had no handlers. SIGINT/SIGTERM now start a bounded graceful shutdown: registered harness children (cursor-agent + bridge) are killed as a tree, cleanup_run_processes runs, the in-flight module is marked interrupted (never running/done) with a module_interrupted trace event, run_status says interrupted, finalizers release the GPU lock and keep the scratch, and the process exits 128+signal (hard deadline 60 s). --resume re-runs interrupted modules and skips done ones; the wave gate treats interrupted as not done. SPEC pages bumped to verified-against 2026-10-28; RUNBOOK gains a section on stopping and resuming a run. --- .../templates/module_rebase.live.prompt.tmpl | 34 +- doc/RUNBOOK-rebase.md | 24 + doc/architecture/SPEC/app/core.md | 10 +- .../SPEC/app/workflow_execution.md | 16 +- doc/architecture/SPEC/engine/lifecycle.md | 72 ++- .../SPEC/engine/steps/rebase_v3.md | 19 +- doc/architecture/SPEC/providers/base.md | 26 +- .../SPEC/providers/claude_code.md | 8 +- doc/architecture/SPEC/providers/codex.md | 8 +- doc/architecture/SPEC/providers/cursor.md | 8 +- doc/architecture/SPEC/rebase_engine.md | 71 ++- doc/architecture/SPEC/run_status.md | 10 +- doc/architecture/SPEC/testing.md | 20 +- doc/architecture/SPEC/tool_bridge.md | 45 +- src/infermatrix_copilot/app/core.py | 5 + .../app/workflow_execution.py | 54 ++- src/infermatrix_copilot/engine/lifecycle.py | 269 ++++++++++- .../engine/steps/rebase_v3.py | 15 +- src/infermatrix_copilot/providers/base.py | 40 ++ .../providers/claude_code.py | 7 +- src/infermatrix_copilot/providers/codex.py | 7 +- src/infermatrix_copilot/providers/cursor.py | 9 +- .../rebase_engine/module_pytest.py | 207 ++++++++- .../rebase_engine/module_rebase.py | 168 ++++++- src/infermatrix_copilot/run_status.py | 22 + src/infermatrix_copilot/testing/runner.py | 36 +- src/infermatrix_copilot/tool_bridge.py | 247 ++++++++-- test/test_adapter_knowledge.py | 11 +- test/test_assembly.py | 55 +++ test/test_kb_intake_gate.py | 2 +- test/test_module_pytest_signals.py | 254 ++++++++++ test/test_run_shutdown.py | 435 ++++++++++++++++++ test/test_tool_bridge_lifecycle.py | 313 +++++++++++++ 33 files changed, 2395 insertions(+), 132 deletions(-) create mode 100644 test/test_module_pytest_signals.py create mode 100644 test/test_run_shutdown.py create mode 100644 test/test_tool_bridge_lifecycle.py diff --git a/adapters/vllm_omni/rebase/templates/module_rebase.live.prompt.tmpl b/adapters/vllm_omni/rebase/templates/module_rebase.live.prompt.tmpl index 4338eccb9..51f52be5d 100644 --- a/adapters/vllm_omni/rebase/templates/module_rebase.live.prompt.tmpl +++ b/adapters/vllm_omni/rebase/templates/module_rebase.live.prompt.tmpl @@ -239,13 +239,35 @@ every later call. Therefore: - Use `imx-omni-pytest …` with `TEST_TIMEOUT_SEC={MODULE_TEST_TIMEOUT_SEC}` for every required verification command. Keep the selected scope, assertions and granted GPU geometry; do not shrink or extend the test budget. -- Never block a tool call on a long wait (`sleep 480 && tail …`). For anything - longer than a short command, launch detached with stdin/stdout/stderr - redirected and a PID file, e.g. - `setsid -f sh -c 'echo $$ > "$1"; shift; exec "$@"' sh "$pidfile" imx-omni-pytest ... "$log" 2>&1` +- Never block a tool call on a long wait (`sleep 480 && tail …`), and never run + a verification command in the foreground of a tool call: a foreground run is + killed at the tool's deadline together with its process group. Launch every + `imx-omni-pytest` DETACHED — own session, stdin from /dev/null, stdout/stderr + to a log, PID saved — in ONE tool call that returns immediately: + ``` + log=$IMX_LOG_DIR/tests/detached-.out; pidfile=$IMX_LOG_DIR/tests/detached-.pid + setsid -f sh -c 'echo $$ > "$1"; shift; exec "$@"' sh "$pidfile" imx-omni-pytest ... "$log" 2>&1 + sleep 1; cat "$pidfile"; ps -o pid,sid,pgid,stat,etimes,cmd -p "$(cat "$pidfile")" + ``` (a plain `nohup ... &` is insufficient: the non-interactive bridge shell waits - for that job). Then poll the saved wrapper PID with short calls, at most one - check per tool call and under 20 seconds each including any sleep. + for that job). The `ps` line is the detach proof: the wrapper's `sid` must + equal its own pid; if it does not, stop and relaunch, never fall back to a + foreground run. The wrapper also re-sessions itself and ignores SIGHUP, so a + tool deadline cannot end it, but it still must not be started in the + foreground, where the tool's output pipe is its stdout. +- Then POLL, at most one check per tool call and under 20 seconds each + including any sleep: `kill -0 "$(cat "$pidfile")"` says live; once it is + gone, read the newest `$IMX_LOG_DIR/tests/outcomes/wrap-*.json` receipt (it is + written by the wrapper at exit) and `tail -n 40 "$log"`. The receipt's + `rc`, `timed_out`, `watchdog_triggered` are the outcome; `signal` and + `signal_origin` say whether a signal ended the child and who sent it. +- A receipt with `signal_origin: "external"` (rc -2/-15/-9, `timed_out` and + `watchdog_triggered` both false, typically ~30 s after start) means a + supervisor OUTSIDE this process tree killed the test — on this host, the + GPU reservation guard, because the GPUs in `CUDA_VISIBLE_DEVICES` were not + reserved for the wrapper's process tree. That is not a test failure and not + your code: record it in debug memory and report the module incomplete with + the receipt path; do not retry in a loop and do not change the test scope. - Guard optional glob or command-substitution results before passing them to stdin-reading commands (`tail`, `head`, `cat`, `grep`); an empty expansion turns `tail -n 5 "$log"` into a bare stdin reader that holds the bridge. diff --git a/doc/RUNBOOK-rebase.md b/doc/RUNBOOK-rebase.md index 9081e9dc2..f74174536 100644 --- a/doc/RUNBOOK-rebase.md +++ b/doc/RUNBOOK-rebase.md @@ -69,6 +69,30 @@ requires a fresh module signal, no typed session/audit refusal, and the required plan decision from that module's current attempt. Plain completion prose does not replace these checks. +### Stopping a run (SIGINT/SIGTERM) and what it leaves behind + +SIGINT (Ctrl-C, or a GPU scheduler's `--timeout`) and SIGTERM are a graceful +shutdown, also when the launcher started the copilot with `nohup … &` and the +process inherited an ignored SIGINT. The engine stops the running harness +session (the vendor CLI and its MCP bridge), kills this run's marked background +processes (including detached test wrappers), releases the GPU lock through the +run finalizers, keeps the upstream scratch, and exits with `128 + signal` +(130 for SIGINT, 143 for SIGTERM) within about a minute; a second signal does +not shorten that. The in-flight module is recorded as `interrupted` in +`substate.json` (never left `running` or promoted to `done`), `run_status.json` +says `interrupted`, and `run_trace.jsonl` carries `module_interrupted` and +`run_interrupted` events. `--resume ` re-runs interrupted modules and +skips completed ones. + +Module test receipts under `/tests/outcomes/wrap-*.json` name the +signal that ended a test and its origin. `signal_origin: "external"` (rc +-2/-15/-9 with `timed_out` and `watchdog_triggered` both false) means a +supervisor outside the run — on a shared host, the GPU reservation guard — +killed the test process: check that the reservation covers the wrapper's +process tree before reading it as a test failure. The bridge's own lifecycle +log, `/bridge/.lifecycle.jsonl`, records why a tool server +exited; a `bridge_unavailable` trace event summarises it per attempt. + ## External checkout pin (EXT1-class guard) Before cutover, the EXTERNAL orchestrator's canonical checkout gains a diff --git a/doc/architecture/SPEC/app/core.md b/doc/architecture/SPEC/app/core.md index c8f17b1ab..e9a9f21b7 100644 --- a/doc/architecture/SPEC/app/core.md +++ b/doc/architecture/SPEC/app/core.md @@ -1,6 +1,6 @@ # app/core.py —— 规范 - + `Copilot` is the headless workflow application. It resolves a TaskSpec, delegates reviewer verdict policy to `review.plan_gate`, confirms when needed, @@ -39,3 +39,11 @@ run（`_contained_run_dir` 拒绝越界路径）；`--repo/--playbook/--task-par 在恢复时是对已保存 task.json 的**断言**而非修改，任一不匹配即 BLOCKED。 发布 launcher 因此能把 resume 绑定到不可变的 campaign，而不会误入更新的 无关任务。损坏的 task.json 同样 BLOCKED 而非崩溃。 + +## 被信号终止的预约 run（2026-10-28） + +`_execute_reserved_locked` 的终态写入（`0 → done`、`BLOCKED_EXIT → blocked`、 +其他 → `failed`）在 `engine.lifecycle.shutdown_requested()` 非空时**跳过**： +`workflow_execution._interrupted` 已经写下 `interrupted`，`128 + signal` 的 +退出码不得被这里改标成 `failed`（那会让 reconciler 把一个可恢复的 run 当作 +失败收尾）。 diff --git a/doc/architecture/SPEC/app/workflow_execution.md b/doc/architecture/SPEC/app/workflow_execution.md index 82df37b73..9892e88ac 100644 --- a/doc/architecture/SPEC/app/workflow_execution.md +++ b/doc/architecture/SPEC/app/workflow_execution.md @@ -1,6 +1,6 @@ # app/workflow_execution.py — workflow execution boundary - + `WorkflowExecution` runs an already selected playbook with the caller's current LLM dependency. It rechecks the repository context before starting, @@ -9,6 +9,20 @@ initializes tracing and notification, seeds executor state with checkout and adapter policy, and drives `Executor` through `run_guarded`. It returns an immutable `ExecutionResult` containing the exit code and blocked reason. +Around `asyncio.run(run_guarded(...))` it installs the run's SIGINT/SIGTERM +handlers (`engine/lifecycle.py::install_run_signal_handlers`) and restores +them afterwards. A signalled run ends through `_interrupted`: run_status +`interrupted` with this process as `child_pid` (`run_status.mark_interrupted`), +a `run_interrupted` trace event (signal, exit code, completed steps), the +per-step marks plus `run : interrupted ()`, and +`ExecutionResult(128 + signal, reason)`. Both paths land there: the +cooperative one (a step saw `shutdown_requested()` and returned blocked, so +`executor.run` returned normally) and the forced one (the run task was +cancelled; `CancelledError`/`KeyboardInterrupt` is caught only when a +shutdown request exists, otherwise it propagates as before). Finalizers have +already run inside `run_guarded` by then: GPU locks are released and the +upstream scratch is kept, so the run is resumable with `--resume`. + A losing run lock leaves active run artifacts untouched. A failed knowledge lock releases an acquired run lock. A caller-provided lock stays owned by the caller. Finalizers and locks run on every executor outcome, including failure. diff --git a/doc/architecture/SPEC/engine/lifecycle.md b/doc/architecture/SPEC/engine/lifecycle.md index 795900a5d..2e19a5341 100644 --- a/doc/architecture/SPEC/engine/lifecycle.md +++ b/doc/architecture/SPEC/engine/lifecycle.md @@ -1,19 +1,36 @@ # engine/lifecycle.py —— 规范 - + -`LOC ~141 · 引擎底座（run 生命周期原语） · refactor-status: ok` +`LOC ~396 · 引擎底座（run 生命周期原语 + 优雅停机） · refactor-status: ok` ## 职责 -executor 自己给不了的两条**进程级**保证：同一 run 的互斥锁，与 run 离开事件 -循环时的 finalizer 挂钩点。 +executor 自己给不了的三条**进程级**保证：同一 run 的互斥锁，run 离开事件 +循环时的 finalizer 挂钩点，以及 SIGINT/SIGTERM 到达时**有界的优雅停机** +（run-20261027-193251：引擎从 `nohup … &` 启动器继承了 SIGINT 的 `SIG_IGN`， +GPU 调度器 `--timeout` 发出的 SIGINT 被整个忽略，随后 reservation guard 逐个 +杀掉其下的测试进程）。 ## 功能 `RunLock` 对 `/.lock` 取排他 advisory `flock`（第二个并发 `--resume` fail fast，而不是交错写进度、共享 checkout）； `register_finalizer`/`finalize` 维护逐 run 的异步 finalizer 表； `run_guarded` 包住 executor 协程，在完成/失败/抛异常的**每条**退出路径上、 -同一事件循环内执行 finalize。 +同一事件循环内执行 finalize，并把自己的 task 绑定给停机路径 +（`bind_run_task`），退出时解绑。 + +停机：`install_run_signal_handlers(run_dir, run_id) -> restorer` 为 SIGINT/ +SIGTERM 安装处理器（**无条件** `signal.signal`——继承来的 `SIG_IGN` 正是要 +覆盖的状态；`asyncio.Runner` 只在默认处理器仍在时才装自己的，所以本处理器 +在 `asyncio.run` 内保持权威）；处理器调用 `request_shutdown(signum, …)` +（幂等：第一次接受，之后只计 `repeats`），它启动守护线程 `imx-shutdown` +并对绑定的 run task 调 `cancel`。工作线程按序：`terminate_children` +（对所有已注册 harness 子进程 `kill_tree`，TERM→宽限→KILL，含其 MCP bridge +后代）→ `cleanup_run_processes(run_dir, run_id)`（带 `IMX_RUN_ID`+`IMX_LOG_DIR` +标记的后台链，含 detached 测试链）→ 注册的 `shutdown hooks`（模块循环用来 +把在飞模块标 `interrupted`）→ 等主线程正常退出事件循环；超过 +`SHUTDOWN_HARD_DEADLINE_S`（60 s）则再杀一次子进程并 `exit_fn(128+signum)` +（默认 `os._exit`）。 ## 公开契约 `RunLock(run_dir).acquire()/release()`（context manager）；`RunLockHeld`； @@ -21,8 +38,22 @@ executor 自己给不了的两条**进程级**保证：同一 run 的互斥锁 executor 在产出之前就抛时为 None）；`finalize(run_dir, outcome)`； `run_guarded(run, run_dir) -> outcome`；`require_file_locking()` 与 `FileLockingUnavailable`（POSIX advisory 锁不可用时抛出的 fail-closed 门）。 -调用方：`cli/copilot.py` 在 run/resume 全程持锁并用 `run_guarded` 包 -`executor.run`；finalizer 注册者是 rebase 流水线（flock 释放、scratch 清理、 +停机面：`ShutdownRequest(signum, run_dir, run_id, requested_at, repeats)` +（`signal_name`、`exit_code = 128 + signum`：SIGINT→130、SIGTERM→143）； +`shutdown_requested() -> ShutdownRequest | None`（协作式轮询点）； +`register_child(proc)/unregister_child(proc)/registered_children()` +（transport 在一次 harness 调用期间注册 Popen，见 `providers/base.py:: +run_harness_cli`）；`register_shutdown_hook(fn)/unregister_shutdown_hook(fn)` +（同步、幂等、不得抛）；`terminate_children(grace_s)`； +`request_shutdown(signum, *, run_dir, run_id, hard_deadline_s, grace_s, exit_fn)`； +`install_run_signal_handlers(run_dir, run_id, *, signals) -> restorer` +（仅主线程；其他线程返回 no-op restorer）；`_reset_shutdown_state_for_tests()`。 +调用方：`app/workflow_execution.py` 在 run/resume 全程持锁、在 +`asyncio.run` 外侧安装/恢复信号处理器并用 `run_guarded` 包 `executor.run`， +捕获 `CancelledError`/`KeyboardInterrupt` 后若 `shutdown_requested()` 非空则走 +`_interrupted` 终局协议（run_status `interrupted`、`run_interrupted` trace、 +退出码 128+signum）；`rebase_engine/module_rebase.py` 在每次 attempt 后轮询 +`shutdown_requested()` 并注册在飞钩子；finalizer 注册者是 rebase 流水线（flock 释放、scratch 清理、 终局报告、CI abort 清理）与 `engine/worktrees.py` 的共享持有释放； `require_file_locking` 由 `engine/worktrees.py::materialize` 和 `idempotency.py`（`key_lock`/`reap_stale`）消费。 @@ -47,13 +78,28 @@ executor 在产出之前就抛时为 None）；`finalize(run_dir, outcome)`； finalize 真正完成，然后才让取消向外传播 —— 光 shield 会让外层先抛、 loop 关闭时把孤儿 finalizer 拦腰取消。 - 什么都没注册时整个挂钩点是 no-op（现有非 rebase playbook 行为不变）。 +- 停机**恰好一次**：第二个信号（任一种）只计 `repeats`，不重启序列、不改 + 退出码；信号处理器只做"记状态 + 起线程 + `call_soon_threadsafe(cancel)`"， + 不在处理器内做 I/O 等待。 +- 协作式与强制式两条路径留下**同一份**持久痕迹：模块 `interrupted`、run + `interrupted`——前者由步骤自己在看到请求后返回 BLOCKED，后者由取消穿过 + finalizer（GPU 锁释放、scratch 保留）后在 `workflow_execution` 收口。 +- 硬截止是兜底不是常态：主线程正常退出事件循环（`_run_loop is None`）时 + 工作线程直接返回，由主线程写终局并返回退出码；只有卡死的 executor 才会 + 被 `os._exit(128+signum)` 终结。 +- 子进程注册面向 **cursor-agent/claude/codex 整棵树**：`subprocess.run` 隐藏 + Popen，所以 transport 必须经 `run_harness_cli` 注册；`terminate_children` + 用 `kill_tree` 走后代，bridge 不会被 reparent 到 init 后继续持有 run 的 + 工具作用域。 ## 边界 —— 不属于这里 不含 teardown 的内容本身（各注册方自带）；不做进度/checkpoint 持久化 （executor）；不是 run 状态查询（`run_status.py`）；不含 step 逻辑。 ## 依赖（允许） -仅 stdlib（asyncio/os/pathlib/fcntl-守卫）。零 engine 内 import。 +stdlib（asyncio/os/signal/subprocess/threading/pathlib/fcntl-守卫）；停机 +工作线程**延迟导入** `testing/process_tree.py::kill_tree` 与 +`rebase_engine/processes.py::cleanup_run_processes`。零 engine 内 import。 ## 扩展点 新的 run 级资源清理 = 调用方去 `register_finalizer`，不改这里；只有新的 @@ -61,10 +107,16 @@ executor 在产出之前就抛时为 None）；`finalize(run_dir, outcome)`； ## 测试 `test_run_lifecycle.py`（同进程/跨进程锁争用、release 后可重取、无 fcntl -降级、finalizer 恰好一次 / 异常与取消隔离 / 失败路径也执行）。 +降级、finalizer 恰好一次 / 异常与取消隔离 / 失败路径也执行）； +`test_run_shutdown.py`（覆盖继承的 SIG_IGN 并恢复、请求幂等与 128+signum、 +工作线程杀注册子进程整棵树后在硬截止 `exit_fn`、无绑定 task 时跑钩子即返回、 +`run_guarded` 绑定/解绑与取消仍 finalize、`run_harness_cli` 注册/超时杀树、 +模块协作式/取消式 `interrupted`、步骤 BLOCKED、wave gate、resume 重跑）。 ## 重构备注 小而承重。`_finalizers` 是模块级进程状态 —— pop-before-run 就是它的自清 机制；若将来单进程要并行多 run，把表挂到 run 对象而不是模块全局。 `run_guarded` 的 shield 循环微妙（取消期间的 teardown 存活是评审加固过的） -—— 改动前先重跑取消路径的测试。 +—— 改动前先重跑取消路径的测试。停机状态（`_shutdown`/`_children`/ +`_shutdown_hooks`/`_run_loop`）同样是模块级进程状态，与"一进程一 run"的 +前提绑定；测试经 `_reset_shutdown_state_for_tests()` 清场。 diff --git a/doc/architecture/SPEC/engine/steps/rebase_v3.md b/doc/architecture/SPEC/engine/steps/rebase_v3.md index de9d5e3b7..27a27ba0b 100644 --- a/doc/architecture/SPEC/engine/steps/rebase_v3.md +++ b/doc/architecture/SPEC/engine/steps/rebase_v3.md @@ -1,6 +1,6 @@ # engine/steps/rebase_v3.py —— 规范 - + `LOC ~2204 · step 库（v3 rebase 装配层） · refactor-status: oversized` @@ -16,8 +16,8 @@ step —— 薄的受治理 wrapper，substate-first、类型化失败、发布 | `rebase.v3_scan` | deterministic/read | `manifest_jobs`；`test_manifest.json` 产物 | | `rebase.v3_wheel` | deterministic/write_workspace | `upstream_commit`（选 commit → 装进目标 venv → **最后**才 pin Dockerfile：装失败绝不留脏树） | | `rebase.v3_assign` | deterministic/read | `active_modules`、`wave1_modules`、`wave2_modules`（path-sync 后按 wave 分派） | -| `rebase.v3_wave_gate` | deterministic/read | wave-1 失败 ⇒ `wave2_modules=[]`；`halt_on_module_failure` ⇒ ESCALATE | -| `rebase.v3_module_rebase` | script/write_workspace | `module__status`（foreach；同 run 串行锁 —— 同一 checkout；substate done/skipped 短路防重入二次执行） | +| `rebase.v3_wave_gate` | deterministic/read | wave-1 失败**或 interrupted** ⇒ `wave2_modules=[]`；`halt_on_module_failure` ⇒ ESCALATE | +| `rebase.v3_module_rebase` | script/write_workspace | `module__status`（foreach；同 run 串行锁 —— 同一 checkout；substate done/skipped 短路防重入二次执行；outcome `interrupted` ⇒ BLOCKED，可 `--resume`） | | `rebase.v3_test_loop` | script/write_workspace | `phase3_failed`；substate `tests.pipeline`/`infra_failures`；空 manifest ⇒ `manifest_empty` | | `rebase.v3_precommit` | script/write_workspace | 无发布 —— substate `tests.precommit`（passed/failed/failed_preexisting/not_declared） | | `rebase.v3_push_gate` | deterministic/read | `push_gate_flagged`、`push_gate_overrides`；不许 ⇒ FORBIDDEN；override 记 trace | @@ -133,3 +133,16 @@ harness 会话通过 MCP tool bridge 拿到同一套工具面，plan gate 在派 - baseline venv（`ensure_baseline_venv`）：硬链接克隆 campaign venv 并换入 baseline commit 的 wheel，让"omni main 对上次对齐的 wheel 是否也失败"可问； 任何失败返回 None，保留保守的回归判定。 + +## 信号中断的模块：BLOCKED 收口、resume 重跑（2026-10-28） + +`_v3_module_rebase` 在 `rebase_module` 返回 `status == "interrupted"` +（引擎收到 SIGINT/SIGTERM，见 `engine/lifecycle` 与 `rebase_engine` 规范）时 +返回 `StepResult(False, BLOCKED, ": interrupted by — resumable with +--resume", state_updates={module__status: interrupted})`：executor 不再推进到 +下一个 foreach 兄弟，run 由 `workflow_execution` 以 `interrupted` 收口。短路 +条件**不变**（只有 done/skipped），因此 `--resume` 重跑 interrupted 模块、跳过 +done；`_v3_wave_gate` 把 interrupted 与 failed 同等视为"wave-1 未完成"（半 +rebase 的模块不是 wave 2 的基础）。测试：`test_assembly.py:: +test_v3_module_rebase_reruns_interrupted_and_blocks_on_interrupt`、 +`test_run_shutdown.py::test_wave_gate_treats_interrupted_as_not_done`。 diff --git a/doc/architecture/SPEC/providers/base.md b/doc/architecture/SPEC/providers/base.md index abc13f3f0..8a3604fdd 100644 --- a/doc/architecture/SPEC/providers/base.md +++ b/doc/architecture/SPEC/providers/base.md @@ -1,8 +1,8 @@ # providers/base.py —— 规范 - + -`LOC ~161 · provider 层契约 + 子进程环境白名单 · refactor-status: ok` +`LOC ~234 · provider 层契约 + 子进程环境白名单 + 受管 CLI 子进程 · refactor-status: ok` ## 职责 每个 provider 都要实现的那组契约，以及决定"厂商 CLI 子进程继承什么环境"的安全原语。 @@ -17,7 +17,8 @@ Settings.tier_target 解析，transport 只做防御性兜底。 --> `ProviderSpec`、`AgentSessionRequest`、`SessionUsage`、`HarnessTransport` （`cli_path`、`require_cli`、`auth_gap`、`run_session`、`complete`）、 -`sanitized_env()`、`flatten_messages()`。 +`sanitized_env()`、`flatten_messages()`、`run_harness_cli(cmd, *, input, cwd, +env, timeout) -> CompletedProcess`。 ## 不变量（**C1**、**C4**、**E2**） - **环境是白名单，不是黑名单**（`_ENV_KEEP` + `LC_`/`XDG_` 前缀）。厂商 CLI 必须保住 @@ -39,7 +40,9 @@ （那是 `Settings`）。 ## 依赖（允许） -stdlib + `..scopes.ToolScope`。它是一个叶子契约模块。 +stdlib + `..scopes.ToolScope`；`run_harness_cli` 内**延迟导入** +`..engine.lifecycle`（`register_child`/`unregister_child`）与 +`..testing.process_tree.kill_tree`。其余仍是叶子契约模块。 ## 扩展点 新的能力标志加进 `ProviderSpec.capabilities`；新的会话上限加进 `AgentSessionRequest`。 @@ -66,3 +69,18 @@ stdlib + `..scopes.ToolScope`。它是一个叶子契约模块。 会话审计的额外**只读**包含根（rebase module agent 必须读上游 checkout 与 目标 venv）。只放宽读包含判定，不放宽写；cursor 传输把它并进 `audit_events(roots=...)`。 + +## 2026-10-28 `run_harness_cli`：受停机路径管理的 harness 子进程 + +`subprocess.run(capture_output=True, text=True, …)` 的等价物，三个 harness +transport（cursor/claude_code/codex）的会话都经它拉起厂商 CLI。与 +`subprocess.run` 的差别，也是它存在的理由： + +- 子进程在调用期间**注册到 run 的停机路径**（`engine.lifecycle.register_child`， + 返回前 `unregister_child`）：引擎收到 SIGINT/SIGTERM 时，harness 会话及其 + MCP bridge 子进程被按序终止，而不是被孤儿化到 init 上继续持有 run 的工具 + 作用域（run-20261027-193251 的第三个缺陷）。 +- 超时时 **杀整棵后代树**（`kill_tree`，TERM 5 s → KILL），而非只给 CLI 领导 + 进程一个 SIGKILL；随后 `TimeoutExpired` 照常抛出，带部分 stdout。 +- `start_new_session=True`：终端的 Ctrl-C 只到引擎，由引擎的处理器按序停会话。 +- `stdin` 在 `input is None` 时为 `/dev/null`，不继承引擎的 stdin。 diff --git a/doc/architecture/SPEC/providers/claude_code.md b/doc/architecture/SPEC/providers/claude_code.md index ee56ed54d..bb62559ef 100644 --- a/doc/architecture/SPEC/providers/claude_code.md +++ b/doc/architecture/SPEC/providers/claude_code.md @@ -1,6 +1,6 @@ # providers/claude_code.py —— 规范 - + `LOC ~205 · harness transport（Claude 订阅） · refactor-status: ok` @@ -59,3 +59,9 @@ x 必须为正。CLI 在每个 API 请求完成**之后**检查预算：它阻 truncated（绝不绿）；refusal/max_budget 另成为 typed refusal，标记文件 不能覆盖。 - `-p json` 解析取最后一个带 `result` 的整行对象，容忍前后杂行。 + +## 2026-10-28 子进程受停机路径管理 + +会话经 `base.run_harness_cli` 拉起 claude CLI（而非 `subprocess.run`）：子进程在调用期间注册到 `engine.lifecycle` 的子进程表，引擎的 +SIGINT/SIGTERM 优雅停机会终止该会话及其 MCP bridge 后代；会话超时杀整棵 +后代树后再抛 `TimeoutExpired`（语义不变：空文本、`truncated`、refusal）。 diff --git a/doc/architecture/SPEC/providers/codex.md b/doc/architecture/SPEC/providers/codex.md index 8f402ad89..a1dace533 100644 --- a/doc/architecture/SPEC/providers/codex.md +++ b/doc/architecture/SPEC/providers/codex.md @@ -1,6 +1,6 @@ # providers/codex.py —— 规范 - + `LOC ~197 · harness transport（ChatGPT 订阅） · refactor-status: ok` @@ -52,3 +52,9 @@ OS 级控制可用的后端；把两者混同，会模糊掉"当前实际生效 ## 2026-09-30 `complete()` 接受并忽略 `max_budget_usd`；`stops_at_spend` 为 False（CLI 无花费阈值）。 + +## 2026-10-28 子进程受停机路径管理 + +会话经 `base.run_harness_cli` 拉起 codex CLI（而非 `subprocess.run`）：子进程在调用期间注册到 `engine.lifecycle` 的子进程表，引擎的 +SIGINT/SIGTERM 优雅停机会终止该会话及其 MCP bridge 后代；会话超时杀整棵 +后代树后再抛 `TimeoutExpired`（语义不变：空文本、`truncated`、refusal）。 diff --git a/doc/architecture/SPEC/providers/cursor.md b/doc/architecture/SPEC/providers/cursor.md index 20218677f..90631c5a5 100644 --- a/doc/architecture/SPEC/providers/cursor.md +++ b/doc/architecture/SPEC/providers/cursor.md @@ -1,6 +1,6 @@ # providers/cursor.py —— 规范 - + `LOC ~207 · harness transport（Cursor 订阅） · refactor-status: ok` @@ -74,3 +74,9 @@ worktree（repo+PR+sha 分键，run queue 不会让同 PR 的两个 run 并发 - `.cursor/mcp.json` 不支持每工具调用超时（CLI 只认 `startupTimeoutSec`）， `--print` 模式也没有对应开关；桥调用的客户端期限由 CLI 决定，prompt 不再 承诺"随会话预算"。 + +## 2026-10-28 子进程受停机路径管理 + +`_run` 经 `base.run_harness_cli` 拉起 cursor-agent（而非 `subprocess.run`）：子进程在调用期间注册到 `engine.lifecycle` 的子进程表，引擎的 +SIGINT/SIGTERM 优雅停机会终止该会话及其 MCP bridge 后代；会话超时杀整棵 +后代树后再抛 `TimeoutExpired`（语义不变：空文本、`truncated`、refusal）。 diff --git a/doc/architecture/SPEC/rebase_engine.md b/doc/architecture/SPEC/rebase_engine.md index c423bdc8a..0e668c3a4 100644 --- a/doc/architecture/SPEC/rebase_engine.md +++ b/doc/architecture/SPEC/rebase_engine.md @@ -1,6 +1,6 @@ # rebase_engine/ —— 规范 - + `LOC ~7500（26 个模块） · repo-rebase-v3 的原生 rebase 引擎 · refactor-status: ok` @@ -25,8 +25,8 @@ | `knowledge_attest.py` | 知识层的 WAL 安全逻辑摘要 + 快照/恢复（只读） | | `knowledge_migrate.py` | PR4d 一次性知识迁移：全锁在手、逐 store 日志化事务 | | `modes.py` | rebase 模式治理：唯一权威解析 + 写回、`mode_*` 旗标、闸冲突裁决 | -| `module_pytest.py` | `imx-omni-pytest` 受闸测试包装（GPU 互斥、看门狗、分层超时） | -| `module_rebase.py` | 单模块 rebase 单元：prompt → agent loop → substate 结果 | +| `module_pytest.py` | `imx-omni-pytest` 受闸测试包装（GPU 互斥、看门狗、分层超时、自建会话、信号回执） | +| `module_rebase.py` | 单模块 rebase 单元：prompt → agent loop → substate 结果；桥存活判定；可中断 | | `parent_compat.py` | 父知识库的只读兼容读取层（`mode=ro`，fail-closed 打开） | | `path_sync.py` | 模块路径图同步 + manifest modules 段重写 + L2 决定应用 | | `phase1_steps.py` | phase-1 组合（归类 + 路径同步），父级报告文件名不变 | @@ -129,7 +129,12 @@ test loop 动作、`precommit_fix`、`RebaseBackends`、LLM client）。 worktree）、`test_assembly.py`（tools/loop/module/prompt/wheel）、 `test_ci_wiring.py`（ci_loop/test_loop/test_manifest）、 `test_ext1_checkout_guard.py`（锁+卫生盾）、`test_knowledge_migrate.py`、 -`test_parent_compat.py`、`test_shell_golden.py`、`test_v3_complete_e2e.py`。 +`test_parent_compat.py`、`test_shell_golden.py`、`test_v3_complete_e2e.py`、 +`test_module_pytest_outcomes.py`、`test_module_pytest_signals.py`（真包装进程： +自建会话、忽略 SIGHUP、SIGTERM 停树并记 `wrapper_signal`、外部 kill 判 +`external` 并写诊断、进程内调用不改调用者会话、子进程 stdin=/dev/null）、 +`test_run_shutdown.py`（模块 `interrupted` 两条路径、钩子幂等）、 +`test_tool_bridge_lifecycle.py`（`bridge_health` 判定与 `bridge_unavailable`）。 ## 重构备注 `ci_loop.py`（1156 行）最大：monitor/分类器/round 编排同居，再长就按 @@ -191,3 +196,61 @@ run-20261024 里这些链在 wave 1 完成后又占着 run 的 GPU 锁约 7 小 的 pytest 项。 - `wheel.py`：`release_venv_holders`（仅 venv 结构存在时、排除调用者祖先链）、 EBUSY/ETXTBSY 重试、baseline rev-parse 前置校验、分支重置前 `git clean -ffd`。 + +## 包装器进程契约：信号来源可追溯、脱离 harness 工具会话（2026-10-28） + +run-20261027-193251 的 `model_executor` 四次模块测试在约 30 s 处结束，回执 +rc -15/-15/-9/-2、`timed_out=false`、`watchdog_triggered=false`——包装器自己 +没有杀它，回执却说不出是谁杀的。证据链（agent 日志第三次 attempt 的说明、 +campaign 日志几百条 "GPU 1 reservation lost"、宿主 root 运行的 +`canhazgpu guard --enforce`）指向宿主 GPU reservation guard 对它视为未预约的 +GPU 进程执行 SIGINT→SIGTERM→SIGKILL 阶梯。`module_pytest.py` 现在： + +- `WrapperProcess.detach()`（**只在真实 CLI 入口**调用：`main(argv=None)`； + 进程内调用者传 argv 时 `process_setup` 默认 False，绝不改调用者的会话）： + 非会话领导者时 `os.setsid()`（`IMX_WRAPPER_FOREGROUND=1` 退出；进程组领导者 + 的 EPERM 记入 `wrapper_detach_error`），并忽略 SIGHUP——harness 工具的前台 + 截止杀进程组、关 stdin，都到不了它。 +- `install_signal_handlers(runner)`：SIGTERM/SIGINT 记入 `received` 并经 + `TestRunner.terminate_active()` 对当前 attempt 的进程组 + 快照后代 + TERM→宽限→KILL；被叫停的包装器绝不把 pytest 留在 GPU 上。`proc.wait()` + 被任何 BaseException 打断时同样先杀树再传播。 +- 回执 `schema_version` 2：新增 `signal`（结束子进程的信号名或 null）、 + `signal_origin` ∈ timeout / watchdog / wrapper_signal / **external** / null、 + `signal_diagnosis`（仅 external：外部监管者杀了测试进程， + `CUDA_VISIBLE_DEVICES` 的 GPU 未为本进程树预约，不是测试失败）、 + `wrapper_pgid/sid/ppid`、`wrapper_session_leader`、`wrapper_detached`、 + `wrapper_detach_error`、`wrapper_parent_comm`、`wrapper_signals_ignored`、 + `wrapper_signals_received`。external 与 wrapper_signal 还在测试日志尾部追加 + "signal postmortem" 页脚（**先于**回执写入，回执的 `log_sha256` 覆盖它）， + 并在 stderr 给出一行结论；stdout/stderr 已关闭时静默。 +- `testing/runner.py::_spawn` 以 `stdin=/dev/null` 拉起测试。 +- live prompt 的 detach 段改为可核验程序：一次工具调用内 `setsid -f` 启动 + + `ps -o pid,sid,…` 证明 `sid == pid`，否则重启而非回退前台；轮询 + `kill -0 $(cat pidfile)`，结束后读 `tests/outcomes/wrap-*.json` 回执； + `signal_origin: "external"` 的含义与处置（记 debug memory、报 incomplete、 + 不循环重试）写进 prompt。 + +## 桥存活判定与可中断的模块循环（2026-10-28） + +- `bridge_health(spec_path, *, since_ts, settle_s=BRIDGE_EXIT_SETTLE_S)` 读 + `tool_bridge` 的 `.lifecycle.jsonl`（只看本次会话开始之后的事件， + start 有而 exit 未到时最多等 2 s——harness 先于其桥子进程返回）并给出 + `verdict` ∈ ok / never_started / exited_early（`detail` 带桥自己的 + `reason`/`rc`）/ no_exit_recorded / restarted，附 `slow_tools`、 + `slowest_tool`。`_harness_attempt` 在非 `ok` 时记 trace + `bridge_unavailable`（step、spec、verdict、detail、计数；不含事件原文）并在 + 结果文本追加一行，供 debug prompt 与 agent 日志读到。 +- `rebase_module` 可中断：每次 attempt 后轮询 `engine.lifecycle. + shutdown_requested()`——有请求则**不再**做 debug 重试，未 done 的模块写 + `{"status": "interrupted", "interrupted_by": , …}`、记 trace + `module_interrupted`、agent 日志加一行；attempt 返回 done 的模块**保持 done**。 + 强制路径：attempt 在飞时 task 被取消（`CancelledError`）同样先标 + `interrupted` 再抛；此外在飞期间注册一个停机钩子（`in_flight` 守卫：attempt + 已返回则让位给协作式判定），工作线程调用它得到同一份痕迹。标记幂等： + 已是 `interrupted` 不重复写、不重复记事件。 +- `rebase_v3._v3_module_rebase` 对 `interrupted` 返回 `StepResult(False, + BLOCKED, …)`（状态更新 `module__status=interrupted`），run 以 interrupted + 收口而不是推进到下一个兄弟；substate 短路**只**接受 done/skipped，所以 + `--resume` 重跑 interrupted 模块、跳过 done；`_v3_wave_gate` 把 interrupted + 与 failed 同等视为未完成。 diff --git a/doc/architecture/SPEC/run_status.md b/doc/architecture/SPEC/run_status.md index 26b2cf92b..b69e6e4c2 100644 --- a/doc/architecture/SPEC/run_status.md +++ b/doc/architecture/SPEC/run_status.md @@ -1,8 +1,8 @@ # run_status.py —— 规范 - + -`LOC ~309 · 持久化的单写者 run 生命周期记录 · refactor-status: ok` +`LOC ~345 · 持久化的单写者 run 生命周期记录 · refactor-status: ok` ## 职责 `run_status.json` —— 一次 Strict run 的**持久、无歧义**的生命周期记录，可跨进程观测。 @@ -17,7 +17,11 @@ server 重启，并且必须能把**崩掉的 run 和正在跑的 run 区分开* `claim_for_execution(run_dir, child_pid)`（**预约 run 子进程的第一步**： 原子 CAS `queued → planning`，同 pid 幂等，输者不得执行）、 `mark_child_started`（**非预约** run 记录自身 pid 的路径）、`mark`、 -`read_status`、`reclaim_queued(run_dir, owner_server_id, +`mark_interrupted(run_dir, *, note, child_pid)`（被信号终止的 run 的子进程侧 +终态：写 `interrupted` **并盖上 `child_pid`**——默认本进程；`child_pid` 为 +null 的 `interrupted` 对 `reclaim_queued` 意味着"预约了、从未拉起"，做过部分 +工作的 run 绝不能被当作没启动过而重新武装；无既有记录的 CLI run 会得到一条 +带 `run_id` 的新记录）、`read_status`、`reclaim_queued(run_dir, owner_server_id, owner_server_pid)`（`interrupted → queued` 重新武装）、 `reconcile_after_wait`/`reconcile_if_dead`/`startup_reconcile`、 `register_server`/`unregister_server`/`server_alive`，以及状态常量 diff --git a/doc/architecture/SPEC/testing.md b/doc/architecture/SPEC/testing.md index fa27d31b0..e35ea2296 100644 --- a/doc/architecture/SPEC/testing.md +++ b/doc/architecture/SPEC/testing.md @@ -1,6 +1,6 @@ # testing/ —— 规范 - + `LOC ~1760（6 个文件） · 测试执行底座（rebase agent shell 测试层的 Python 移植） · refactor-status: ok` @@ -30,7 +30,9 @@ test_runner.sh 的 Python 移植。仓库专属数据（watchdog 模式、artifa `cleanup_orphan_gpu_procs`、`wait_gpu_memory_idle`。 - `process_tree`：`collect_descendants`、`kill_tree(pids, identity=...)`、`kill_by_pattern`。 - `runner`：`TestRunner.run(job, env, baseline, dry_run) -> TestOutcome`、 - `TestJob`、`RunPlan`、`TestOutcome`、`strip_cov_flags`、`cleanup_test_artifacts`、 + `TestRunner.terminate_active() -> bool`（对正在跑的 attempt 的进程组 + 快照 + 后代 TERM→宽限→KILL；无 attempt 在跑返回 False；可自信号处理器/他线程调用， + `_spawn` 的 `proc.wait()` 随后返回子进程的信号 rc）、`TestJob`、`RunPlan`、`TestOutcome`、`strip_cov_flags`、`cleanup_test_artifacts`、 `log_file_for`/`pass_marker_for`、`TIMEOUT_RC=124`、`PY_TIMEOUT_MARGIN_SEC`。 - `watchdog`：`WatchdogPatterns(.from_yaml)`、`LogWatchdog(.start/.stop/.check_once/.result)`、`WatchdogResult`。 - `watchdog_learn`：`record`、`harvest`、`eligible_patterns`、`promote`、 @@ -81,7 +83,8 @@ env 前缀 → `scrub_agent_shell_env(extra_safe_prefixes=…)`（manifest 数 ## 测试 `test_testing_substrate.py`（77 个，覆盖全部六模块）、`test_shell_golden.py` -（TestRunner 对 shell 的 command-echo 金样）；`watchdog_learn` 另由 +（TestRunner 对 shell 的 command-echo 金样）、`test_module_pytest_signals.py` +（经包装器：`terminate_active` 停树、stdin=/dev/null）；`watchdog_learn` 另由 `test_curator.py`（harvest exactly-once）与 `test_knowledge_foundation.py` （record 身份字段、撕尾修复）钉住。 @@ -90,3 +93,14 @@ env 前缀 → `scrub_agent_shell_env(extra_safe_prefixes=…)`（manifest 数 接线挤在一个方法里，密度高但每行都背着一条已记录的竞态教训 —— 拆分前先读 注释里的事故史。身份校验逻辑分居 `_record_walk`（记录侧）与 `kill_tree` （消费侧），契约靠注释维系；若再演化，考虑一个共享的 identity 小模块。 + +## 子进程 stdin 与可被叫停的 attempt（2026-10-28） + +- `_spawn` 以 `stdin=subprocess.DEVNULL` 拉起 `bash --norc -c`：测试绝不继承 + 包装器的 stdin（harness 工具在截止时关闭的管道、或会把挂断传给交互式子进程 + 的终端）。 +- `_spawn` 期间 `self._active_kill` 指向本 attempt 的树杀手（进程组 + 身份 + 快照），`terminate_active()` 调它；`proc.wait()` 被 BaseException（包装器被 + KeyboardInterrupt/SystemExit 展开）打断时先刷新快照、杀树，再让异常传播—— + 包装器被展开时测试树不得继续占 GPU。消费方是 + `rebase_engine/module_pytest.py::WrapperProcess` 的 SIGTERM/SIGINT 处理器。 diff --git a/doc/architecture/SPEC/tool_bridge.md b/doc/architecture/SPEC/tool_bridge.md index d84413ebd..a4423df1e 100644 --- a/doc/architecture/SPEC/tool_bridge.md +++ b/doc/architecture/SPEC/tool_bridge.md @@ -1,8 +1,8 @@ # tool_bridge.py —— 规范 - + -`LOC ~308 · 面向 harness 会话的 scoped 工具 MCP server · refactor-status: ok` +`LOC ~783 · 面向 harness 会话的 scoped 工具 MCP server · refactor-status: ok` ## 职责 以 stdio MCP server 的形式，把本次 run 的 scoped 工具暴露给 harness 会话 —— @@ -16,7 +16,11 @@ ## 公开契约 `write_bridge_spec(...)`、`load_bridge_spec(path)`、`make_dispatcher(...)`、 -`build_server(spec_path)`、`main(argv)`。 +`build_server(spec_path, lifecycle=None)`、`run_bridge(spec_path) -> rc`、 +`main(argv)`；存活面：`BridgeLifecycle(path).record(kind, **fields)`、 +`lifecycle_log_path(spec_path)`、`read_lifecycle_events(spec_path)`、 +`timed_call(call, lifecycle, slow_after_s)`、`_offload(fn)`、 +`install_bridge_signal_dispositions()`、`SLOW_TOOL_CALL_S`（60 s）。 ## 不变量（**C1**、**C2**、**E2**） - **每一次内置工具调用都经过 `tools.dispatch`**，带着该 step 反序列化出来的 @@ -41,7 +45,10 @@ stdlib + `mcp`（`[mcp]` extra）+ `..tools` + `..scopes` + `..run_trace` + knowledge/repo-map 工厂。 ## 测试 -`test_tool_bridge.py`。 +`test_tool_bridge.py`；`test_tool_bridge_lifecycle.py`（offload 保签名且在工作 +线程跑、慢调用不阻塞并发调用（真 FastMCP server）、慢调用记录、真 stdio +进程忽略 SIGINT/SIGHUP 且 stdin 关闭后 rc 0 并记 `exit(stdin_closed)`、 +`bridge_health` 判定矩阵、`_harness_attempt` 记 `bridge_unavailable`）。 ## 重构备注 **并非每个后端都够得到这个 server**：`providers/deepseek.py` 跑在 harness 原生 bash 上， @@ -68,3 +75,33 @@ run 的 `CUDA_VISIBLE_DEVICES`，也没有 `HF_HOME`——`_agent_shell_env` 与 看到的是整机 GPU。spec 的 `rebase.host_env`（仅这两个非机密键，str→str， 否则 `TypeError`）在重建后端前写进桥进程 env。`rebase.target_venv` 同理 携带已解析的目标 venv；凭据与完整子环境仍不进 spec。 + +## 存活：工具下线到工作线程 + 生命周期日志（2026-10-28） + +run-20261027-193251：`infermatrix-tools` 在第一次 attempt 的中途"discovery +failed"（`record_debug_memory` 不可用），此前同一会话里已成功调用 27 次。 +根因：桥里**每个**工具都是同步 handler，而 FastMCP 对同步 handler 是在事件 +循环上**内联**调用的（`call_fn_with_arg_validation` 直接 `fn(**args)`），所以 +一次 `request_plan_review`（上限 `strict_backend_timeout_s`=1800 s）或 600 s 的 +`run_shell` 会让整个 server 在那几分钟里不应答 ping、不应答任何其他调用； +cursor-agent 的每调用截止先到期，待排队的调用在它返回时集中冲刷，随后 +client 把 server 判为失败，余下数小时不再路由到它。桥进程本身是否死掉， +以前没有任何记录可查。 + +- **所有** handler 经 `_offload` 注册为 `async def` 包装（`functools.wraps` + 保留原签名——FastMCP 由签名推 input schema——内部 + `anyio.to_thread.run_sync`），事件循环在长调用期间保持响应； + `test_slow_tool_does_not_block_a_concurrent_call` 用真 server 证明。 +- `run_bridge` 为每个 spec 写 `/bridge/.lifecycle.jsonl` + （与 `bridge_trace.jsonl` 分离：后者记工具调用、全 run 共享；前者记**这个 + 进程**）：`start`（pid/ppid/sid/pgid/spec/argv）、`slow_tool` + （≥ `SLOW_TOOL_CALL_S` 的调用：tool/elapsed_s，是 client 截止切断的唯一旁证）、 + `exit`（`reason` ∈ stdin_closed / signal / system_exit / exception / + import_error / build_error，`rc`、`calls`、`parent_alive`）。写失败绝不抛。 +- 进程对 SIGINT/SIGHUP/SIGPIPE 置 `SIG_IGN`（`install_bridge_signal_dispositions`， + 在 `run_bridge` **最先**执行——SDK 导入在高负载宿主上要数秒，期间到达的信号 + 不得先杀掉还没开始的 server）：stdio server 的寿命就是它的 stdin，终端 Ctrl-C + 只应到引擎、由引擎按序停会话；harness 关闭 stdin 是唯一正常退出。 + 与 `mcp_server.py`/`thin_mcp_server.py` 的 SIGINT 忽略同一原则。 +- 消费方：`rebase_engine/module_rebase.py::bridge_health` 在会话结束后读该 + 日志并分类（见 rebase_engine 规范），非 `ok` 时记 `bridge_unavailable`。 diff --git a/src/infermatrix_copilot/app/core.py b/src/infermatrix_copilot/app/core.py index 98a3f47d5..4d71f7424 100644 --- a/src/infermatrix_copilot/app/core.py +++ b/src/infermatrix_copilot/app/core.py @@ -454,6 +454,11 @@ def _execute_reserved_locked(self, run_dir: Path, lock: RunLock, # nothing to distinguish a stale-head refusal from a missing tier or a # failed checkout — the reason was previously printed to the child's # console and discarded. + from ..engine.lifecycle import shutdown_requested + if shutdown_requested() is not None: + # a signalled run already wrote `interrupted`; `128 + signal` + # must not be re-labelled as a failure here + return code rs.mark(run_dir, {0: rs.DONE, BLOCKED_EXIT: rs.BLOCKED}.get(code, rs.FAILED), note=self.last_blocked_reason if code == BLOCKED_EXIT else "") return code diff --git a/src/infermatrix_copilot/app/workflow_execution.py b/src/infermatrix_copilot/app/workflow_execution.py index d38198da8..0650d35a8 100644 --- a/src/infermatrix_copilot/app/workflow_execution.py +++ b/src/infermatrix_copilot/app/workflow_execution.py @@ -9,7 +9,9 @@ from ..adapters.base import AdapterError from ..config import Settings from ..engine.executor import Executor -from ..engine.lifecycle import RunLock, RunLockHeld, run_guarded +from ..engine.lifecycle import (RunLock, RunLockHeld, + install_run_signal_handlers, run_guarded, + shutdown_requested) from ..engine.registry import StepRegistry from ..metrics import format_metrics_line from ..notify import BLOCKED_EXIT, Notifier @@ -124,8 +126,27 @@ def run( # run_guarded finalizes inside the event loop: playbooks that # register run finalizers (lifecycle.register_finalizer) get # teardown on every exit path. Nothing registered == no-op. - outcome = asyncio.run( - run_guarded(executor.run(playbook, state), run_dir)) + # SIGINT/SIGTERM (a GPU scheduler ending the reservation, an + # operator) become a bounded graceful shutdown: harness children + # and marked background processes are stopped, the run task is + # cancelled through the same finalizers, and the run is left + # resumable and marked `interrupted` — never `running`. + restore = install_run_signal_handlers(run_dir, run_dir.name) + try: + outcome = asyncio.run( + run_guarded(executor.run(playbook, state), run_dir)) + except (asyncio.CancelledError, KeyboardInterrupt): + req = shutdown_requested() + if req is None: + raise + return self._interrupted(run_dir, trace, req) + finally: + restore() + req = shutdown_requested() + if req is not None: + # the cooperative path: a step noticed the request and + # returned blocked before cancellation reached it + return self._interrupted(run_dir, trace, req, outcome) if outcome.status == "done": notifier.resolve() @@ -152,3 +173,30 @@ def run( knowledge_lock.release() if held_lock is None: lock.release() + + @staticmethod + def _interrupted(run_dir: Path, trace: RunTrace, req, + outcome=None) -> ExecutionResult: + """Terminal protocol for a signalled run: durable `interrupted` + status (run_status.json names this process as the child that ran, + so no reconciler mistakes it for a never-launched reservation), a + trace event, the operator-facing line, and `128 + signal`.""" + from .. import run_status as rs + + reason = (f"interrupted by {req.signal_name} — run is resumable " + f"(--resume {run_dir.name})") + try: + rs.mark_interrupted(run_dir, note=reason) + except OSError: + pass + trace.record("run_interrupted", signal=req.signal_name, + exit_code=req.exit_code, + steps_completed=sorted( + k for k, r in (getattr(outcome, "step_results", None) + or {}).items() if r.ok)) + for step_id, r in (getattr(outcome, "step_results", None) or {}).items(): + mark = style("✓", "green") if r.ok else style("✗", "red", "bold") + print(f" {mark} {step_id}: {r.summary}") + print(f"run {run_dir.name}: interrupted ({req.signal_name}) ({run_dir})") + print(style(" ⚠ ", "yellow", "bold") + reason) + return ExecutionResult(req.exit_code, reason) diff --git a/src/infermatrix_copilot/engine/lifecycle.py b/src/infermatrix_copilot/engine/lifecycle.py index 52cf8ef85..7afdd85e4 100644 --- a/src/infermatrix_copilot/engine/lifecycle.py +++ b/src/infermatrix_copilot/engine/lifecycle.py @@ -12,12 +12,27 @@ registered (every current playbook) the hook is a no-op; the rebase pipeline registers teardown (locks, scratch clones, store flushes) here so that resume, crash, and success all release resources the same way. +* graceful shutdown — `install_run_signal_handlers` turns SIGINT/SIGTERM into + a bounded, ordered teardown of the run (harness children first, then the + run's marked background processes, then the run's own finalizers via task + cancellation), with a hard `os._exit` deadline so the process cannot hang + on a child that ignores signals. A GPU scheduler ends a reservation with + SIGINT; a launcher started with `nohup ... &` hands the process an + INHERITED `SIG_IGN` for SIGINT, which Python keeps — so a run that does not + install its own handler is simply unkillable by the scheduler + (run-20261027-193251: `kill -INT` was a no-op; SIGTERM then orphaned the + harness session and the detached test chain on the GPUs). """ from __future__ import annotations import asyncio import os +import signal +import subprocess +import threading +import time +from dataclasses import dataclass, field from pathlib import Path from typing import Any, Awaitable, Callable @@ -126,16 +141,256 @@ async def run_guarded(run: Awaitable[Any], run_dir: Path) -> Any: shielded future until teardown actually completes, then let the cancellation propagate.""" outcome: Any = None + task = asyncio.current_task() + if task is not None: # a signal cancels THIS task, unwinding the executor + bind_run_task(asyncio.get_running_loop(), task) try: outcome = await run return outcome finally: - fin = asyncio.ensure_future(finalize(run_dir, outcome)) - cancelled = False - while not fin.done(): - try: - await asyncio.shield(fin) - except asyncio.CancelledError: - cancelled = True # teardown still owed; keep waiting + try: + fin = asyncio.ensure_future(finalize(run_dir, outcome)) + cancelled = False + while not fin.done(): + try: + await asyncio.shield(fin) + except asyncio.CancelledError: + cancelled = True # teardown still owed; keep waiting + finally: + _unbind_run_task() if cancelled: raise asyncio.CancelledError() + + +# ── graceful shutdown ───────────────────────────────────────────────────────── +# +# Process-wide by design: one CLI process runs one run, and a signal is a +# process-level event. Everything here is idempotent so the cooperative path +# (the module loop notices the request and records `interrupted`) and the +# forced path (task cancellation + hard deadline) may both run. + +# Grace the shutdown worker gives harness children between SIGTERM and SIGKILL, +# and the hard ceiling after which the process `os._exit`s whatever is still +# pending. Both are deliberately short: the scheduler that sent the signal is +# about to reclaim the GPUs regardless. +CHILD_TERM_GRACE_S = 15.0 +SHUTDOWN_HARD_DEADLINE_S = 60.0 + + +@dataclass +class ShutdownRequest: + """The one shutdown this process accepted: which signal, when, and the + run it belongs to. Later signals of either kind are recorded but do not + restart the sequence.""" + signum: int + run_dir: Path | None = None + run_id: str = "" + requested_at: float = field(default_factory=time.time) + repeats: int = 0 + + @property + def signal_name(self) -> str: + try: + return signal.Signals(self.signum).name + except ValueError: # pragma: no cover - unknown number + return f"signal {self.signum}" + + @property + def exit_code(self) -> int: + """Shell convention: 128 + signal (130 for SIGINT, 143 for SIGTERM).""" + return 128 + int(self.signum) + + +_shutdown: ShutdownRequest | None = None +_shutdown_lock = threading.Lock() +_children: dict[int, subprocess.Popen] = {} +_shutdown_hooks: list[Callable[[ShutdownRequest], None]] = [] +_run_loop: tuple[asyncio.AbstractEventLoop, asyncio.Task] | None = None +_previous_handlers: dict[int, Any] = {} + + +def shutdown_requested() -> ShutdownRequest | None: + """The accepted shutdown request, or None while the run is undisturbed. + Long-running loops poll this between units of work (cooperative path).""" + return _shutdown + + +def register_child(proc: subprocess.Popen) -> None: + """Make a harness child (cursor-agent and its MCP bridge, claude, codex) + reachable by the shutdown worker. `subprocess.run` hides the Popen, so + transports spawn with Popen and register it for the call's duration.""" + with _shutdown_lock: + _children[proc.pid] = proc + + +def unregister_child(proc: subprocess.Popen) -> None: + with _shutdown_lock: + _children.pop(proc.pid, None) + + +def registered_children() -> list[int]: + with _shutdown_lock: + return sorted(_children) + + +def register_shutdown_hook(fn: Callable[[ShutdownRequest], None]) -> None: + """A synchronous hook the shutdown worker calls once (after the children + are stopped). The module loop registers "record the in-flight module as + interrupted" here so the forced path leaves the same durable mark as the + cooperative one. Hooks must be idempotent and must never raise.""" + with _shutdown_lock: + if fn not in _shutdown_hooks: + _shutdown_hooks.append(fn) + + +def unregister_shutdown_hook(fn: Callable[[ShutdownRequest], None]) -> None: + with _shutdown_lock: + if fn in _shutdown_hooks: + _shutdown_hooks.remove(fn) + + +def bind_run_task(loop: asyncio.AbstractEventLoop, task: asyncio.Task) -> None: + """Tell the shutdown path which task to cancel: `run_guarded` binds its + own task so a signal unwinds the executor through the finalizers.""" + global _run_loop + _run_loop = (loop, task) + + +def _unbind_run_task() -> None: + global _run_loop + _run_loop = None + + +def terminate_children(*, grace_s: float = CHILD_TERM_GRACE_S) -> list[int]: + """SIGTERM every registered child and its descendants, wait `grace_s`, + SIGKILL the survivors. Returns pids still alive afterwards. The + descendant walk matters: cursor-agent's MCP bridge is its child and + would otherwise be reparented to init with the run's tool scope.""" + from ..testing.process_tree import kill_tree + + with _shutdown_lock: + procs = list(_children.values()) + pids = [p.pid for p in procs if p.poll() is None] + if not pids: + return [] + return kill_tree(pids, term_grace=grace_s, kill_grace=2.0) + + +def _shutdown_worker(req: ShutdownRequest, *, hard_deadline_s: float, + grace_s: float, exit_fn: Callable[[int], None]) -> None: + """Ordered teardown off the main thread: children → marked processes → + hooks. Then wait for the main thread to unwind normally; past the hard + deadline, exit the process regardless (a wedged executor thread must not + turn a SIGINT into an immortal process).""" + started = time.monotonic() + try: + terminate_children(grace_s=grace_s) + except Exception: # noqa: BLE001 - teardown never aborts on one failure + pass + if req.run_dir is not None and req.run_id: + try: + from ..rebase_engine.processes import cleanup_run_processes + cleanup_run_processes(req.run_dir, req.run_id) + except Exception: # noqa: BLE001 + pass + with _shutdown_lock: + hooks = list(_shutdown_hooks) + for fn in hooks: + try: + fn(req) + except Exception: # noqa: BLE001 - a hook must not stop the others + pass + remaining = hard_deadline_s - (time.monotonic() - started) + deadline = time.monotonic() + max(0.0, remaining) + while time.monotonic() < deadline: + if _run_loop is None: # the run left the event loop normally + return + time.sleep(0.2) + try: + terminate_children(grace_s=1.0) + except Exception: # noqa: BLE001 + pass + exit_fn(req.exit_code) + + +def request_shutdown(signum: int, *, run_dir: Path | None = None, + run_id: str = "", hard_deadline_s: float = SHUTDOWN_HARD_DEADLINE_S, + grace_s: float = CHILD_TERM_GRACE_S, + exit_fn: Callable[[int], None] = os._exit) -> ShutdownRequest: + """Accept a shutdown (idempotent): record it, start the worker thread and + cancel the bound run task. Safe to call from a signal handler — it only + sets state, starts a thread and schedules a thread-safe callback.""" + global _shutdown + with _shutdown_lock: + if _shutdown is not None: + _shutdown.repeats += 1 + return _shutdown + req = ShutdownRequest(signum=signum, run_dir=run_dir, run_id=run_id) + _shutdown = req + threading.Thread( + target=_shutdown_worker, args=(req,), + kwargs={"hard_deadline_s": hard_deadline_s, "grace_s": grace_s, + "exit_fn": exit_fn}, + name="imx-shutdown", daemon=True).start() + bound = _run_loop + if bound is not None: + loop, task = bound + try: + loop.call_soon_threadsafe(task.cancel) + except RuntimeError: # loop already closed + pass + return req + + +def install_run_signal_handlers(run_dir: Path, run_id: str, *, + signals: tuple[int, ...] = (signal.SIGINT, signal.SIGTERM), + ) -> Callable[[], None]: + """Install SIGINT/SIGTERM handlers for a run and return the restorer. + + `signal.signal` is used unconditionally: an inherited `SIG_IGN` (nohup / + non-interactive `&` launchers) is exactly the state this must override, + and `asyncio.Runner` only installs its own SIGINT handler when the + default one is still in place, so ours stays authoritative inside + `asyncio.run`. Only the main thread may install handlers; elsewhere this + is a no-op restorer.""" + if threading.current_thread() is not threading.main_thread(): + return lambda: None + + def _handler(signum, _frame): + req = request_shutdown(signum, run_dir=run_dir, run_id=run_id) + if req.repeats == 0: + try: + print(f"\n[run] {signal.Signals(signum).name} received — " + "stopping the harness session and this run's background " + f"processes (resumable: --resume {run_id})", flush=True) + except (OSError, ValueError): # stdout gone with the launcher + pass + return req + + previous: dict[int, Any] = {} + for signum in signals: + try: + previous[signum] = signal.signal(signum, _handler) + except (ValueError, OSError): # pragma: no cover - unsupported signal + continue + _previous_handlers.update(previous) + + def _restore() -> None: + for signum, handler in previous.items(): + try: + signal.signal(signum, handler) + except (ValueError, OSError): # pragma: no cover + pass + _previous_handlers.pop(signum, None) + + return _restore + + +def _reset_shutdown_state_for_tests() -> None: + """Test seam: forget any accepted request, children and hooks.""" + global _shutdown, _run_loop + with _shutdown_lock: + _shutdown = None + _children.clear() + _shutdown_hooks.clear() + _run_loop = None diff --git a/src/infermatrix_copilot/engine/steps/rebase_v3.py b/src/infermatrix_copilot/engine/steps/rebase_v3.py index c8a362afe..b1ebe31e1 100644 --- a/src/infermatrix_copilot/engine/steps/rebase_v3.py +++ b/src/infermatrix_copilot/engine/steps/rebase_v3.py @@ -1220,9 +1220,11 @@ async def _v3_wave_gate(ctx: StepContext) -> StepResult: sub = _substate(ctx) data = sub.read() wave1 = ctx.state.get("wave1_modules") or [] + # `interrupted` counts as not-done here: a half-rebased wave-1 module is + # no base for wave 2 either (a resumed run re-runs it before this gate) failed = [m for m in wave1 if ((data.get("modules") or {}).get(m) or {}) - .get("status") == "failed"] + .get("status") in ("failed", "interrupted")] if not failed: return StepResult(True, summary="wave 1 clean; wave 2 proceeds") if _task_params(ctx).get("halt_on_module_failure"): @@ -2093,6 +2095,17 @@ async def _v3_module_rebase(ctx: StepContext) -> StepResult: run_dir=ctx.run_dir), module_test_plan=test_plan, trace=ctx.trace) + if outcome["status"] == "interrupted": + # a SIGINT/SIGTERM stopped the agent mid-module: the step BLOCKS so + # the run ends (status interrupted, scratch kept) instead of the + # executor advancing to the next sibling; resume re-runs the module + # because the short-circuit above accepts only done/skipped + return StepResult(False, FailureKind.BLOCKED, + f"{module}: interrupted by " + f"{outcome.get('interrupted_by', 'signal')} — " + "resumable with --resume", + outputs={"state_updates": { + f"module_{module}_status": "interrupted"}}) return StepResult(True, summary=f"{module}: {outcome['status']} " f"(debug_attempts={outcome['debug_attempts']})", diff --git a/src/infermatrix_copilot/providers/base.py b/src/infermatrix_copilot/providers/base.py index 8055fcf6f..fb03bd5bf 100644 --- a/src/infermatrix_copilot/providers/base.py +++ b/src/infermatrix_copilot/providers/base.py @@ -176,6 +176,46 @@ def complete(self, *, system: str, messages: list[dict], raise NotImplementedError +def run_harness_cli(cmd: list[str], *, input: str | None, cwd: str | None, + env: dict[str, str] | None, timeout: float | None + ) -> "subprocess.CompletedProcess[str]": + """`subprocess.run(capture_output=True, text=True, ...)` for a harness CLI, + with two properties `subprocess.run` lacks: + + - the child is REGISTERED with the run's shutdown path for the duration + of the call, so SIGINT/SIGTERM to the engine terminates the harness + session (and its MCP bridge child) instead of orphaning it onto init; + - a timeout kills the whole descendant tree, not just the leader + (`subprocess.run` sends one SIGKILL to the CLI and leaves the bridge). + + The child gets its own session: a terminal Ctrl-C then reaches the engine + only, whose handler stops the session in order. Timeout semantics match + `subprocess.run`: `TimeoutExpired` carries the partial stdout.""" + import subprocess + + from ..engine.lifecycle import register_child, unregister_child + from ..testing.process_tree import kill_tree + + proc = subprocess.Popen( + cmd, stdin=subprocess.PIPE if input is not None else subprocess.DEVNULL, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, env=env, + text=True, encoding="utf-8", errors="replace", start_new_session=True) + register_child(proc) + try: + try: + stdout, stderr = proc.communicate(input, timeout=timeout) + except subprocess.TimeoutExpired: + kill_tree([proc.pid], term_grace=5.0, kill_grace=2.0) + try: + proc.communicate(timeout=10) + except (subprocess.TimeoutExpired, OSError, ValueError): + pass + raise + finally: + unregister_child(proc) + return subprocess.CompletedProcess(cmd, proc.returncode, stdout, stderr) + + def flatten_messages(system: str, messages: list[dict]) -> str: """Render an internal block-protocol conversation as one prompt string for harnesses without a messages channel. Tool-less conversations only (the diff --git a/src/infermatrix_copilot/providers/claude_code.py b/src/infermatrix_copilot/providers/claude_code.py index 0b645a148..cbaa1d271 100644 --- a/src/infermatrix_copilot/providers/claude_code.py +++ b/src/infermatrix_copilot/providers/claude_code.py @@ -35,6 +35,7 @@ HarnessTransport, SessionUsage, flatten_messages, + run_harness_cli, sanitized_env, ) from .registry import PROVIDERS @@ -87,10 +88,8 @@ def _run(self, prompt_text: str, *, system: str, cwd: str, cmd += ["--mcp-config", str(mcp_config), "--strict-mcp-config", "--allowedTools", f"mcp__{_BRIDGE_SERVER}"] try: - proc = subprocess.run( - cmd, input=prompt_text, cwd=cwd, env=sanitized_env(), - capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=timeout_s, check=False) + proc = run_harness_cli(cmd, input=prompt_text, cwd=cwd, + env=sanitized_env(), timeout=timeout_s) stdout = proc.stdout or "" except subprocess.TimeoutExpired as exc: raw = exc.stdout or b"" diff --git a/src/infermatrix_copilot/providers/codex.py b/src/infermatrix_copilot/providers/codex.py index c72f8c402..393e56e85 100644 --- a/src/infermatrix_copilot/providers/codex.py +++ b/src/infermatrix_copilot/providers/codex.py @@ -32,6 +32,7 @@ HarnessTransport, SessionUsage, flatten_messages, + run_harness_cli, sanitized_env, ) from .registry import PROVIDERS @@ -122,10 +123,8 @@ def _run(self, text: str, *, cwd: str, timeout_s: float, model: str = "", cmd += ["-"] timed_out = False try: - proc = subprocess.run( - cmd, input=text, cwd=cwd, env=sanitized_env(), - capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=timeout_s, check=False) + proc = run_harness_cli(cmd, input=text, cwd=cwd, + env=sanitized_env(), timeout=timeout_s) stdout = proc.stdout or "" except subprocess.TimeoutExpired as exc: timed_out = True diff --git a/src/infermatrix_copilot/providers/cursor.py b/src/infermatrix_copilot/providers/cursor.py index f94fa3c08..2c1bfcb46 100644 --- a/src/infermatrix_copilot/providers/cursor.py +++ b/src/infermatrix_copilot/providers/cursor.py @@ -33,6 +33,7 @@ HarnessTransport, SessionUsage, flatten_messages, + run_harness_cli, sanitized_env, ) from .registry import PROVIDERS @@ -145,10 +146,10 @@ def _run(self, text: str, *, cwd: str, timeout_s: float, timed_out = False self._last_exit = (0, "") try: - proc = subprocess.run( - cmd, input=text, cwd=cwd, env=sanitized_env(), - capture_output=True, text=True, encoding="utf-8", - errors="replace", timeout=timeout_s, check=False) + # registered with the run's shutdown path: a SIGINT/SIGTERM to + # the engine stops this session (and its bridge child) in order + proc = run_harness_cli(cmd, input=text, cwd=cwd, + env=sanitized_env(), timeout=timeout_s) stdout = proc.stdout or "" self._last_exit = (int(proc.returncode or 0), (proc.stderr or "")[-400:].strip()) diff --git a/src/infermatrix_copilot/rebase_engine/module_pytest.py b/src/infermatrix_copilot/rebase_engine/module_pytest.py index de6e0debf..1389e56f1 100644 --- a/src/infermatrix_copilot/rebase_engine/module_pytest.py +++ b/src/infermatrix_copilot/rebase_engine/module_pytest.py @@ -20,7 +20,26 @@ `imx-omni-pytest python -c '...'` runs that command under the same gating (the parent wrapper accepted both shapes). An adapter may declare ``rebase.testing.module_test_platform`` so broad module runs do not execute -tests that CI routes exclusively to a different accelerator platform.""" +tests that CI routes exclusively to a different accelerator platform. + +Process contract (run-20261027-193251, four module test runs killed at +~30 s with rc -2/-15/-9 and neither timeout nor watchdog set): the wrapper +is launched from a harness tool shell with a foreground deadline, inside a +session it does not own, on a host where a GPU reservation guard signals +processes it considers unreserved. The wrapper therefore + + * puts itself in its OWN session (`os.setsid`, unless it already leads + one or `IMX_WRAPPER_FOREGROUND=1` opts out) and ignores SIGHUP, so the + tool's deadline kill of its process group and the loss of its stdin do + not end the test; + * handles SIGTERM/SIGINT by stopping the test tree (TERM → grace → KILL) + and recording the signal — a wrapper told to stop never leaves pytest + on a GPU; + * records in the outcome receipt which signal ended the child and who + sent it (`signal_origin`: timeout | watchdog | wrapper_signal | + external) plus its own pid/pgid/sid/ppid context, so a kill by a + supervisor outside this process tree is distinguishable from the + wrapper's own enforcement in the receipt alone.""" from __future__ import annotations @@ -29,8 +48,10 @@ import os import re import shlex +import signal import sys import tempfile +import time from pathlib import Path from ..testing.runner import TestJob, TestOutcome, TestRunner @@ -38,6 +59,124 @@ _ACCELERATOR_PLATFORMS = frozenset({"cuda", "musa", "npu", "rocm", "xpu"}) +RECEIPT_SCHEMA_VERSION = 2 # 2: signal / signal_origin / process-context fields + + +class WrapperProcess: + """What this wrapper process did to itself and what reached it: the + receipt's process-context block. `detach()` is called once by the real + CLI entry (never by in-process tests, which must not re-session the + test runner); `install_signal_handlers()` records SIGTERM/SIGINT and + stops the active test tree through the runner.""" + + def __init__(self) -> None: + self.detached = False + self.detach_error = "" + self.ignored: list[str] = [] + self.received: list[dict] = [] + self._runner: TestRunner | None = None + + # -- setup --------------------------------------------------------------- + def detach(self) -> None: + if os.environ.get("IMX_WRAPPER_FOREGROUND") == "1": + self.detach_error = "IMX_WRAPPER_FOREGROUND=1" + elif not hasattr(os, "setsid"): + self.detach_error = "setsid unavailable" + else: + try: + if os.getsid(0) == os.getpid(): + self.detached = True # already a session leader + else: + os.setsid() + self.detached = True + except OSError as exc: + # a process-group leader cannot setsid (job-control shell + # foreground): stay, and say so in the receipt + self.detach_error = f"setsid: {exc.strerror or exc}" + for signum in (getattr(signal, "SIGHUP", None),): + if signum is None: + continue + try: + signal.signal(signum, signal.SIG_IGN) + self.ignored.append(signal.Signals(signum).name) + except (OSError, ValueError): + pass + + def install_signal_handlers(self, runner: TestRunner) -> None: + self._runner = runner + for signum in (signal.SIGTERM, signal.SIGINT): + try: + signal.signal(signum, self._on_signal) + except (OSError, ValueError): + pass + + def _on_signal(self, signum, _frame) -> None: + name = signal.Signals(signum).name + self.received.append({"signal": name, "ts": time.time()}) + runner = self._runner + if runner is not None: + try: + runner.terminate_active() + except Exception: # noqa: BLE001 — a handler must return + pass + + # -- receipt ------------------------------------------------------------- + @staticmethod + def _parent_comm(ppid: int) -> str | None: + try: + return Path(f"/proc/{ppid}/comm").read_text( + encoding="utf-8").strip()[:32] + except OSError: + return None + + def context(self) -> dict: + pid = os.getpid() + ppid = os.getppid() + sid = os.getsid(0) if hasattr(os, "getsid") else None + pgid = os.getpgrp() if hasattr(os, "getpgrp") else None + return { + "wrapper_pgid": pgid, "wrapper_sid": sid, "wrapper_ppid": ppid, + "wrapper_session_leader": sid == pid, + "wrapper_detached": self.detached, + "wrapper_detach_error": self.detach_error or None, + "wrapper_parent_comm": self._parent_comm(ppid), + "wrapper_signals_ignored": list(self.ignored), + "wrapper_signals_received": [r["signal"] for r in self.received], + } + + def classify(self, outcome: TestOutcome) -> dict: + """`signal` (name of the signal that ended the child, or None) and + `signal_origin`: who sent it. `external` is the diagnostic case — + the child died from a signal the wrapper's timeout, watchdog and + own handlers never sent, i.e. an outside supervisor targeted the + test process (on a shared GPU host, the reservation guard).""" + rc = outcome.rc + name = None + if isinstance(rc, int) and rc < 0: + try: + name = signal.Signals(-rc).name + except ValueError: + name = f"SIG{-rc}" + if outcome.timed_out: + origin = "timeout" + elif outcome.watchdog_triggered: + origin = "watchdog" + elif self.received: + origin = "wrapper_signal" + elif name is not None: + origin = "external" + else: + origin = None + diagnosis = None + if origin == "external": + diagnosis = (f"child killed by {name} that neither the timeout, " + "the watchdog nor this wrapper sent: an external " + "supervisor targeted the test process (on a shared " + "GPU host, a reservation guard — the GPUs in " + "CUDA_VISIBLE_DEVICES were not reserved for this " + "process tree); the run itself did not fail") + return {"signal": name, "signal_origin": origin, + "signal_diagnosis": diagnosis} def _needs_gpu_lock(args: list[str]) -> bool: @@ -103,7 +242,8 @@ def identity(st): def _write_outcome_receipt(log_dir: Path, *, args: list[str], run_id: str, attempt: str, timeout: int, cuda: str, outcome: TestOutcome, effective_args: list[str], - platform_marker: str | None) -> None: + platform_marker: str | None, + process: WrapperProcess | None = None) -> None: """Atomic per-attempt runner outcome, without commands, paths, or env dumps. Requested argv is not effective inner argv/scope proof (pytest defaults @@ -112,7 +252,8 @@ def _write_outcome_receipt(log_dir: Path, *, args: list[str], run_id: str, """ starttime = None try: - # Linux self identity only; no other process or environment is read. + # Linux self identity; the only other process read is the parent's + # comm (`WrapperProcess.context`), never its environment. stat = Path("/proc/self/stat").read_text(encoding="utf-8") starttime = int(stat.rsplit(")", 1)[1].split()[19]) except (OSError, ValueError, IndexError): @@ -126,8 +267,10 @@ def _write_outcome_receipt(log_dir: Path, *, args: list[str], run_id: str, r"-?\d+|GPU-[0-9a-fA-F-]+|MIG-[0-9a-fA-F-]+(?:/\d+/\d+)?", part ) for part in cuda_subset) log_sha, log_status = _log_hash(outcome.log_file) + process = process or WrapperProcess() receipt = { - "schema_version": 1, + "schema_version": RECEIPT_SCHEMA_VERSION, + **process.context(), **process.classify(outcome), "run_id": (run_id if re.fullmatch(r"run-\d{8}-\d{6}-[0-9a-f]{6}", run_id) else None), "run_id_sha256": hashlib.sha256(run_id.encode()).hexdigest(), @@ -160,8 +303,17 @@ def _write_outcome_receipt(log_dir: Path, *, args: list[str], run_id: str, Path(temporary).unlink(missing_ok=True) -def main(argv: list[str] | None = None) -> int: +def main(argv: list[str] | None = None, *, + process_setup: bool | None = None) -> int: + """`process_setup` (default: only for the real CLI, i.e. `argv is None`) + re-sessions this process and installs its signal handlers — an + in-process caller (tests) must never have ITS session changed.""" args = list(sys.argv[1:] if argv is None else argv) + process = WrapperProcess() + if process_setup is None: + process_setup = argv is None + if process_setup: + process.detach() repo = os.environ.get("IMX_TARGET_REPO", "") log_dir = os.environ.get("IMX_LOG_DIR", "") if not repo or not log_dir: @@ -282,16 +434,53 @@ def review_fn(test_name: str, snippet: str) -> str: # pass), while gpu_lock=True holds the lock for the run job = TestJob(key=_job_key(args), command=command, timeout_sec=timeout, min_gpus=0, gpu_lock=_needs_gpu_lock(args), index=0) + if process_setup: + process.install_signal_handlers(runner) outcome = runner.run(job, dict(os.environ)) + verdict = process.classify(outcome) + if verdict["signal_diagnosis"] or process.received: + # before the receipt: its log hash must cover the footer + _append_signal_footer(outcome.log_file, verdict, process) _write_outcome_receipt(Path(log_dir), args=args, run_id=run_ident, attempt=_attempt, timeout=timeout, cuda=cuda, outcome=outcome, effective_args=effective_args, - platform_marker=platform_marker) - if outcome.skipped: - print(f"imx-omni-pytest: skipped — {outcome.skip_reason}", - file=sys.stderr) + platform_marker=platform_marker, process=process) + try: + if outcome.skipped: + print(f"imx-omni-pytest: skipped — {outcome.skip_reason}", + file=sys.stderr) + if verdict["signal_diagnosis"]: + print(f"module test wrapper: {verdict['signal_diagnosis']}", + file=sys.stderr) + elif process.received: + print("module test wrapper: stopped by " + + ", ".join(r["signal"] for r in process.received) + + f" (child rc={outcome.rc})", file=sys.stderr) + except (BrokenPipeError, OSError): + pass # the launching shell is gone; the receipt and log remain return outcome.rc +def _append_signal_footer(log_file: str, verdict: dict, + process: WrapperProcess) -> None: + """Postmortem footer for a signal-ended child: the debug agent reads the + log, not the receipt, and must not mistake an external kill for a test + failure.""" + if not log_file: + return + lines = ["", "=== module test wrapper signal postmortem ===", + f"signal={verdict['signal']} origin={verdict['signal_origin']}", + "wrapper_signals_received=" + + (",".join(r["signal"] for r in process.received) or "none"), + f"wrapper_pid={os.getpid()} detached={process.detached}"] + if verdict["signal_diagnosis"]: + lines.append(verdict["signal_diagnosis"]) + try: + with open(log_file, "a", encoding="utf-8") as stream: + stream.write("\n".join(lines) + "\n") + except OSError: + pass + + if __name__ == "__main__": # pragma: no cover sys.exit(main()) diff --git a/src/infermatrix_copilot/rebase_engine/module_rebase.py b/src/infermatrix_copilot/rebase_engine/module_rebase.py index 4d568b90a..00b9fd994 100644 --- a/src/infermatrix_copilot/rebase_engine/module_rebase.py +++ b/src/infermatrix_copilot/rebase_engine/module_rebase.py @@ -10,10 +10,13 @@ from __future__ import annotations +import asyncio from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping +from ..engine.lifecycle import (ShutdownRequest, register_shutdown_hook, + shutdown_requested, unregister_shutdown_hook) from ..run_trace import RunTrace from ..scopes import ToolScope from .agent_loop import GATED_TOOL_NAMES, run_agent_loop @@ -216,8 +219,6 @@ async def _harness_attempt(prompt: str, *, module: str, config, bypass). Returns the same dict shape the in-process attempt does so `rebase_module`'s retry/debug logic is untouched. """ - import asyncio - from ..providers import AgentSessionRequest from ..providers.registry import transport_for_id from ..tool_bridge import write_bridge_spec @@ -306,6 +307,17 @@ def signal_snapshot(path): started = _time.time() outcome = await asyncio.to_thread(transport.run_session, req) + bridge = (bridge_health(spec_path, since_ts=started) if spec_path + else {"verdict": "ok", "detail": ""}) + if bridge["verdict"] != "ok" and trace is not None: + # the harness only ever reports "tool not found"; the bridge's own + # lifecycle log says whether it was never launched, died, or was + # starved by a slow call (run-20261027-193251: discovery lost + # mid-session after a 30-minute synchronous plan review) + trace.record("bridge_unavailable", step=f"rebase.module.{module}", + spec=str(spec_path), **{k: v for k, v in bridge.items() + if k != "events"}) + native, pre_gate = _native_writes(scope.root, run_dir, before, started) if native and trace is not None: trace.record("harness_native_writes", step=f"rebase.module.{module}", @@ -347,6 +359,11 @@ def signal_snapshot(path): count=len(audit_reads), notes=audit_reads[:10]) result_text += ("\nSession read audit (not a completion veto): " + "; ".join(audit_reads[:3])) + if bridge["verdict"] != "ok": + result_text += ("\nTool bridge " + bridge["verdict"].replace("_", " ") + + (f" ({bridge['detail']})" if bridge.get("detail") else "") + + ": the harness lost or never had the infermatrix-tools " + "server; see bridge/*.lifecycle.jsonl.") if not done: reasons = [] if not signaled: @@ -368,6 +385,61 @@ def signal_snapshot(path): "plan_done": plan_done} +BRIDGE_EXIT_SETTLE_S = 2.0 # the harness returns before its bridge child finishes exiting + + +def bridge_health(spec_path: Path, *, since_ts: float, + settle_s: float = BRIDGE_EXIT_SETTLE_S) -> dict: + """Read the bridge process's lifecycle log for one session and classify + it. `verdict` is one of: + + - `ok`: started once, exited because the harness closed stdin; + - `never_started`: no bridge launched after `since_ts` — the harness + never connected (discovery failed from the first turn); + - `exited_early`: the bridge left while the harness was still running + (signal, exception, SystemExit) — `detail` carries its own reason; + - `no_exit_recorded`: started, but no exit event appeared within + `settle_s` of the session ending (SIGKILL, or still alive: orphaned); + - `restarted`: more than one start — the harness relaunched it. + + `slow_tools` counts calls ≥ `tool_bridge.SLOW_TOOL_CALL_S`: the usual + cause of a client-side deadline that makes a harness drop the server.""" + import time as _time + + from ..tool_bridge import read_lifecycle_events + + events: list[dict] = [] + deadline = _time.monotonic() + max(0.0, settle_s) + while True: + events = [e for e in read_lifecycle_events(spec_path) + if float(e.get("ts") or 0) >= since_ts - 1.0] + starts = [e for e in events if e.get("kind") == "start"] + exits = [e for e in events if e.get("kind") == "exit"] + if not starts or exits or _time.monotonic() >= deadline: + break + _time.sleep(0.1) + slow = [e for e in events if e.get("kind") == "slow_tool"] + report = {"starts": len(starts), "exits": len(exits), "slow_tools": len(slow), + "slowest_tool": max(slow, key=lambda e: e.get("elapsed_s", 0)) + .get("tool") if slow else None, "detail": "", "events": events} + if not starts: + verdict = "never_started" + elif len(starts) > 1: + verdict = "restarted" + elif not exits: + verdict = "no_exit_recorded" + else: + last = exits[-1] + if last.get("reason") == "stdin_closed" and not last.get("parent_alive"): + verdict = "ok" + else: + verdict = "exited_early" + report["detail"] = (f"reason={last.get('reason')} rc={last.get('rc')} " + f"{last.get('detail') or ''}").strip() + report["verdict"] = verdict + return report + + def _append_agent_log(path: str, text: str) -> None: with Path(path).open("a", encoding="utf-8") as log: log.write(text) @@ -457,26 +529,80 @@ async def _attempt(p: str, *, require_plan_review: bool = True) -> dict: from .processes import cleanup_run_processes cleanup_run_processes(Path(config.log_dir), substate.run_id) - # parent parity: an incomplete first run gets up to max_debug_retries - # follow-up attempts with the debug prompt built from the failure text. - # The plan gate PERSISTS across attempts: once the module's decision was - # written, debug retries run unlocked (the debug prompt carries no - # plan-review contract, and the parent's re-locking left debug agents - # unable to edit — a fixed, recorded parent defect); an initial run that - # never passed the gate keeps it closed for retries. + # A SIGINT/SIGTERM to the engine must leave this module `interrupted` + # (never `running`, which resume would mistake for a live sibling, nor + # `done`): the cooperative path checks after each attempt; the forced + # path (task cancelled, or the shutdown worker's hook racing the + # unwind) writes the same mark. Idempotent by construction. debug_attempts = 0 - result = await _attempt(prompt) - gate_passed = bool(result.get("plan_done")) - while not result.get("done") and \ - debug_attempts < config.max_debug_retries: - debug_attempts += 1 - debug_prompt = build_debug_prompt( - module, result.get("text", ""), - prompt_data.debug_prompt_template_live - or prompt_data.debug_prompt_template, "") - result = await _attempt(debug_prompt, - require_plan_review=not gate_passed) - gate_passed = gate_passed or bool(result.get("plan_done")) + in_flight = False # True only while an attempt is awaited (the hook's guard) + + def _mark_interrupted(req: ShutdownRequest | None) -> dict: + signal_name = req.signal_name if req is not None else "cancelled" + current = (substate.read().get("modules") or {}).get(module) or {} + if current.get("status") == "interrupted": + return current + outcome = {"status": "interrupted", "exit_code": -1, + "debug_attempts": debug_attempts, "turns": 0, + "interrupted_by": signal_name, + "summary": (f"interrupted by {signal_name} while the module " + f"agent was running; resume re-runs it")} + substate.update({"modules": {module: outcome}}) + if trace is not None: + trace.record("module_interrupted", step=f"rebase.module.{module}", + signal=signal_name, debug_attempts=debug_attempts) + try: + _append_agent_log(agent_log, f"\nModule interrupted by {signal_name}; " + "resume re-runs this module.\n") + except OSError: + pass + return outcome + + def _hook(req: ShutdownRequest) -> None: + # the worker thread's view: only an attempt that is still running + # is interrupted; one that already returned is judged by the + # cooperative path below (a `done` result stays done) + if in_flight: + _mark_interrupted(req) + + async def _guarded_attempt(p: str, **kw) -> dict: + nonlocal in_flight + in_flight = True + try: + return await _attempt(p, **kw) + finally: + in_flight = False + + register_shutdown_hook(_hook) + try: + # parent parity: an incomplete first run gets up to max_debug_retries + # follow-up attempts with the debug prompt built from the failure + # text. The plan gate PERSISTS across attempts: once the module's + # decision was written, debug retries run unlocked (the debug prompt + # carries no plan-review contract, and the parent's re-locking left + # debug agents unable to edit — a fixed, recorded parent defect); an + # initial run that never passed the gate keeps it closed for retries. + result = await _guarded_attempt(prompt) + gate_passed = bool(result.get("plan_done")) + while not result.get("done") and shutdown_requested() is None and \ + debug_attempts < config.max_debug_retries: + debug_attempts += 1 + debug_prompt = build_debug_prompt( + module, result.get("text", ""), + prompt_data.debug_prompt_template_live + or prompt_data.debug_prompt_template, "") + result = await _guarded_attempt(debug_prompt, + require_plan_review=not gate_passed) + gate_passed = gate_passed or bool(result.get("plan_done")) + except asyncio.CancelledError: + _mark_interrupted(shutdown_requested()) + raise + finally: + unregister_shutdown_hook(_hook) + + req = shutdown_requested() + if req is not None and not result.get("done"): + return _mark_interrupted(req) outcome = {"status": "done" if result.get("done") else "failed", "exit_code": 0 if result.get("done") else -1, diff --git a/src/infermatrix_copilot/run_status.py b/src/infermatrix_copilot/run_status.py index b1d764e48..5098070f7 100644 --- a/src/infermatrix_copilot/run_status.py +++ b/src/infermatrix_copilot/run_status.py @@ -193,6 +193,28 @@ def mark(run_dir: str | Path, state: str, *, note: str = "") -> dict: return _locked_update(run_dir, lambda _cur: upd) # type: ignore[return-value] +def mark_interrupted(run_dir: str | Path, *, note: str = "", + child_pid: Optional[int] = None) -> dict: + """Child-side `interrupted` on a signalled run (SIGINT from a GPU + scheduler, SIGTERM from an operator). Stamps `child_pid` (this process by + default): an `interrupted` record with a null child pid means "reserved, + never launched" to `reclaim_queued`, and a run that did partial work must + never be re-armed as if it had not started. A CLI run without a prior + record gets one, so `--resume` tooling and reconcilers see the same + terminal truth the MCP path writes.""" + pid = int(child_pid if child_pid is not None else os.getpid()) + + def _upd(cur: dict) -> dict: + upd: dict[str, Any] = {"state": INTERRUPTED, "child_pid": pid} + if cur.get("run_id") is None: + upd["run_id"] = Path(run_dir).name + if note: + upd["note"] = note + return upd + + return _locked_update(run_dir, _upd) # type: ignore[return-value] + + # ── liveness ────────────────────────────────────────────────────────────────── def pid_alive(pid: Optional[int]) -> bool: """True if `pid` names a live process. `os.kill(pid, 0)` probes without diff --git a/src/infermatrix_copilot/testing/runner.py b/src/infermatrix_copilot/testing/runner.py index eb6cb7b56..01894d98f 100644 --- a/src/infermatrix_copilot/testing/runner.py +++ b/src/infermatrix_copilot/testing/runner.py @@ -323,6 +323,23 @@ def __init__(self, *, repo_root: Path, tests_dir: Path, lambda: len(visible_devices(self.cuda))) self.watchdog_interval = watchdog_interval self.notify_download = notify_download + # the active attempt's tree killer, set for the duration of `_spawn`: + # a wrapper signal handler calls `terminate_active()` so the test + # tree never outlives a wrapper that was told to stop + self._active_kill: Callable[[], None] | None = None + self._active_lock = threading.Lock() + + def terminate_active(self) -> bool: + """TERM → grace → KILL the attempt currently running (process group + plus every snapshotted descendant). Returns False when nothing is + running. Safe from a signal handler or another thread; `_spawn`'s + `proc.wait()` then returns the child's signal rc.""" + with self._active_lock: + kill = self._active_kill + if kill is None: + return False + kill() + return True def run(self, job: TestJob, env: dict[str, str], *, baseline: bool = False, dry_run: bool = False) -> TestOutcome: @@ -500,8 +517,12 @@ def _spawn(self, exec_cmd: str, job: TestJob, env: dict[str, str], # shell at SHLVL=0. That can put a foreign CLI ahead of the # explicitly selected runtime. Keep setup and test environments # authoritative; explicit sourcing in a job still works. + # stdin is /dev/null: a test must never inherit the wrapper's + # stdin (a harness tool pipe that closes at the tool's deadline, + # or a terminal whose hangup would reach an interactive child) proc = subprocess.Popen(["bash", "--norc", "-c", exec_cmd], cwd=self.repo_root, env=env, + stdin=subprocess.DEVNULL, stdout=lf, stderr=lf, start_new_session=True) # capture the pgid while the leader is certainly alive: the final @@ -551,6 +572,9 @@ def _snap_loop(): def _kill_snapshot(): self._terminate_tree(proc.pid, pgid, snapshot=tree["idmap"]) + with self._active_lock: + self._active_kill = _kill_snapshot + watchdog = None if self.patterns is not None: watchdog = LogWatchdog( @@ -574,8 +598,18 @@ def safety(): # only matters if the primary path wedged t_primary.start() t_safety.start() try: - rc = proc.wait() + try: + rc = proc.wait() + except BaseException: + # the WRAPPER is being unwound (KeyboardInterrupt, a + # SystemExit from a handler): the tree must not outlive + # it holding GPUs — kill, then let the unwind continue + _snapshot() + self._terminate_tree(proc.pid, pgid, snapshot=tree["idmap"]) + raise finally: + with self._active_lock: + self._active_kill = None stop_snap.set() snap_thread.join(timeout=10) # cancel then JOIN: a fired primary timer is mid-escalation in diff --git a/src/infermatrix_copilot/tool_bridge.py b/src/infermatrix_copilot/tool_bridge.py index 3eb51d839..ba0d50db4 100644 --- a/src/infermatrix_copilot/tool_bridge.py +++ b/src/infermatrix_copilot/tool_bridge.py @@ -27,14 +27,28 @@ Those tools can propose knowledge candidates, and opening a cross-process write path for them was declined — a harness session may read this repo's knowledge, never add to it. + +Liveness (run-20261027-193251): every bridged tool is a SYNC handler, and +FastMCP calls sync handlers inline on its event loop, so one long call (a +30-minute plan review, a 10-minute `run_shell`) froze the whole server — no +pings, no other tool, and the harness marked the server undiscoverable for +the rest of a multi-hour session. Handlers therefore run on worker threads +(`_offload`), the process ignores SIGINT/SIGHUP like the other stdio servers, +and a per-spec lifecycle log under `/bridge/` records start, slow +calls and the exit reason so the module loop can report `bridge_unavailable` +instead of letting the agent find out by tool failure. """ from __future__ import annotations import argparse +import functools import json import os import re +import signal +import sys +import time from pathlib import Path from .run_trace import RunTrace @@ -322,6 +336,109 @@ def _fn_from_schema(name: str, schema: dict, call): return fn +# -- liveness ---------------------------------------------------------------- +SLOW_TOOL_CALL_S = 60.0 # a call this long is logged: it is what starves a client deadline + + +def lifecycle_log_path(spec_path: Path) -> Path: + """`/bridge/.lifecycle.jsonl` next to the spec.""" + spec_path = Path(spec_path) + return spec_path.with_name(f"{spec_path.stem}.lifecycle.jsonl") + + +class BridgeLifecycle: + """Append-only record of this bridge PROCESS: `start`, `slow_tool`, + `exit` (with the reason). Separate from `bridge_trace.jsonl`, which + records tool calls and is shared by every bridge of the run; this file + answers "was the server alive, and why did it stop" per spec. Writes + never raise — a logging failure must not take the server down.""" + + def __init__(self, path: Path): + self.path = Path(path) + self.calls = 0 + + def record(self, kind: str, **fields) -> None: + event = {"ts": time.time(), "kind": kind, "pid": os.getpid(), + "ppid": os.getppid(), **fields} + try: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a", encoding="utf-8") as stream: + stream.write(json.dumps(event, default=str) + "\n") + except OSError: + pass + + +def read_lifecycle_events(spec_path: Path) -> list[dict]: + """Parsed lifecycle events for a spec (missing/corrupt lines ignored).""" + out: list[dict] = [] + try: + for line in lifecycle_log_path(spec_path).read_text( + encoding="utf-8").splitlines(): + line = line.strip() + if not line: + continue + try: + event = json.loads(line) + except ValueError: + continue + if isinstance(event, dict): + out.append(event) + except OSError: + pass + return out + + +def _parent_alive(ppid: int) -> bool: + """True while the launching harness process exists (reparenting to init + or a reaper means it is gone).""" + if ppid <= 1 or os.getppid() != ppid: + return False + try: + os.kill(ppid, 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def _offload(fn): + """Run a sync tool handler on a worker thread so the MCP event loop keeps + answering pings, list_tools and other calls during a long handler. The + wrapper keeps the original signature (FastMCP derives the input schema + from it) and is a coroutine function, which FastMCP awaits.""" + import anyio + + @functools.wraps(fn) + async def _async(*args, **kwargs): + return await anyio.to_thread.run_sync( + functools.partial(fn, *args, **kwargs)) + + return _async + + +def timed_call(call, lifecycle: BridgeLifecycle | None, + slow_after_s: float = SLOW_TOOL_CALL_S): + """Count calls and log the slow ones (≥ `slow_after_s`) to the lifecycle + file: a client-side deadline that cut a call off shows up here as the + only call longer than the deadline, which is otherwise invisible.""" + if lifecycle is None: + return call + + def bound(name: str, args: dict): + started = time.monotonic() + lifecycle.calls += 1 + try: + return call(name, args) + finally: + elapsed = time.monotonic() - started + if elapsed >= slow_after_s: + lifecycle.record("slow_tool", tool=name, + elapsed_s=round(elapsed, 1)) + + return bound + + # -- server ------------------------------------------------------------------ def make_dispatcher(scope: ToolScope, roots: tuple[str, ...], trace: RunTrace, extra: dict | None = None, gate: "PlanGate | None" = None): @@ -386,9 +503,11 @@ def bound(name: str, args: dict): return bound -def build_server(spec_path: Path): +def build_server(spec_path: Path, lifecycle: BridgeLifecycle | None = None): """Build the FastMCP server for one spec. Import of the MCP SDK is local - so the module stays importable (spec read/write) without the extra.""" + so the module stays importable (spec read/write) without the extra. + Every handler is registered OFFLOADED (see `_offload`): the event loop + must stay responsive during a long tool call.""" from mcp.server.fastmcp import FastMCP scope, spec = load_bridge_spec(spec_path) @@ -407,8 +526,9 @@ def build_server(spec_path: Path): or ("edit_file", "run_pytest", "run_precommit")), trace=trace) - _call = traced_call(make_dispatcher(scope, roots, trace, - extra=rebase_extra or None, gate=gate), spec) + _call = timed_call(traced_call(make_dispatcher( + scope, roots, trace, extra=rebase_extra or None, gate=gate), spec), + lifecycle) mcp = FastMCP("infermatrix-tool-bridge") # `dispatch` resolves `extra` BEFORE the builtin registry, so a name the @@ -420,38 +540,42 @@ def build_server(spec_path: Path): _name = _d.get("name", "") if not _name or _name not in rebase_extra: continue - mcp.add_tool(_fn_from_schema(_name, _d.get("input_schema") or {}, - _call), - name=_name, description=_d.get("description", "")) + mcp.add_tool(_offload(_fn_from_schema( + _name, _d.get("input_schema") or {}, _call)), + name=_name, description=_d.get("description", "")) + + def tool(name: str): + return lambda fn: mcp.add_tool(_offload(fn), name=name, + description=TOOLS[name].description) if "read_file" in allowed: - @mcp.tool(description=TOOLS["read_file"].description) + @tool("read_file") def read_file(path: str, offset: int = 0) -> str: return _call("read_file", {"path": path, "offset": offset}) if "list_dir" in allowed: - @mcp.tool(description=TOOLS["list_dir"].description) + @tool("list_dir") def list_dir(path: str) -> str: return _call("list_dir", {"path": path}) if "grep" in allowed: - @mcp.tool(description=TOOLS["grep"].description) + @tool("grep") def grep(pattern: str, path: str, regex: bool = False) -> str: return _call("grep", {"pattern": pattern, "path": path, "regex": regex}) if "write_file" in allowed: - @mcp.tool(description=TOOLS["write_file"].description) + @tool("write_file") def write_file(path: str, content: str) -> str: return _call("write_file", {"path": path, "content": content}) if "edit_file" in allowed: - @mcp.tool(description=TOOLS["edit_file"].description) + @tool("edit_file") def edit_file(path: str, old: str, new: str) -> str: return _call("edit_file", {"path": path, "old": old, "new": new}) if "run_shell" in allowed: - @mcp.tool(description=TOOLS["run_shell"].description) + @tool("run_shell") def run_shell(cmd: str, cwd: str = "") -> str: return _call("run_shell", {"cmd": cmd, "cwd": cwd or None}) @@ -478,23 +602,27 @@ def _call(name: str, args: dict) -> str: raise RuntimeError(str(out.get("error") or "tool error")) return str(out["result"]) - @mcp.tool(description=tools["diff_stat"].description) + def tool(name: str): + return lambda fn: mcp.add_tool(_offload(fn), name=name, + description=tools[name].description) + + @tool("diff_stat") def diff_stat() -> str: return _call("diff_stat", {}) - @mcp.tool(description=tools["file_at_base"].description) + @tool("file_at_base") def file_at_base(path: str, offset: int = 0) -> str: return _call("file_at_base", {"path": path, "offset": offset}) - @mcp.tool(description=tools["show_commit"].description) + @tool("show_commit") def show_commit(sha: str) -> str: return _call("show_commit", {"sha": sha}) - @mcp.tool(description=tools["search_history"].description) + @tool("search_history") def search_history(term: str, path: str = "") -> str: return _call("search_history", {"term": term, "path": path}) - @mcp.tool(description=tools["calc"].description) + @tool("calc") def calc(expr: str) -> str: return _call("calc", {"expr": expr}) @@ -544,8 +672,12 @@ def _register_knowledge_tools(mcp, spec: dict, scope: ToolScope, effect=f"doc tools unavailable: {type(exc).__name__}: {exc}") return - @mcp.tool(description="Search the curated knowledge base (general + this " - "repo's slice); returns matching doc paths.") + def tool(description: str): + return lambda fn: mcp.add_tool(_offload(fn), name=fn.__name__, + description=description) + + @tool("Search the curated knowledge base (general + this repo's slice); " + "returns matching doc paths.") def doc_search(query: str, limit: int = 20) -> str: trace.record("tool_call", tool="doc_search", ok=True, out_of_scope=False, path=None) @@ -554,8 +686,8 @@ def doc_search(query: str, limit: int = 20) -> str: f"{h.get('path')}:{h.get('line')}:{str(h.get('text') or '').strip()}" for h in hits) or "(no matches)" - @mcp.tool(description="Read a knowledge doc returned by doc_search " - "(paged; pass the previous next_offset).") + @tool("Read a knowledge doc returned by doc_search (paged; pass the " + "previous next_offset).") def doc_read(path: str, offset: int = 0) -> str: trace.record("tool_call", tool="doc_read", ok=True, out_of_scope=False, path=path) @@ -571,9 +703,9 @@ def doc_read(path: str, offset: int = 0) -> str: effect=f"repo_map unavailable: {type(exc).__name__}: {exc}") return if "repo_map" in map_tools: - tool = map_tools["repo_map"] + tool_def = map_tools["repo_map"] - @mcp.tool(description=tool.description) + @tool(tool_def.description) def repo_map(query: str) -> str: # dispatch with extra= mirrors the in-process extra-tool path # (traced, bypasses the builtin allowlist by design) @@ -584,22 +716,69 @@ def repo_map(query: str) -> str: return str(out["result"]) +def install_bridge_signal_dispositions() -> None: + """A stdio MCP server lives exactly as long as its stdin: it must NOT die + on a terminal Ctrl-C (SIGINT goes to the whole foreground group, and the + engine's handler stops the harness in order), on a hangup of a terminal + it never used, or on a write to a closed pipe (surfaces as an exception + the transport handles). The harness closing stdin is the one exit.""" + for signum in (signal.SIGINT, signal.SIGHUP, signal.SIGPIPE): + try: + signal.signal(signum, signal.SIG_IGN) + except (OSError, ValueError, AttributeError): + pass + + +def run_bridge(spec_path: Path) -> int: + """Serve one spec over stdio until stdin closes, logging start/exit (with + the reason) to the lifecycle file. The log is the module loop's evidence + for a `bridge_unavailable` trace event after the session.""" + # dispositions FIRST: a signal that lands while the SDK is still importing + # (seconds on a loaded host) must not end the server before it starts + install_bridge_signal_dispositions() + lifecycle = BridgeLifecycle(lifecycle_log_path(spec_path)) + ppid = os.getppid() + lifecycle.record("start", spec=str(spec_path), argv=sys.argv[1:], + sid=os.getsid(0) if hasattr(os, "getsid") else None, + pgid=os.getpgrp() if hasattr(os, "getpgrp") else None) + try: + server = build_server(spec_path, lifecycle) + except ImportError: + lifecycle.record("exit", reason="import_error", rc=1, + detail="the MCP SDK is not installed") + sys.stderr.write("the tool bridge needs the MCP SDK — " + "pip install 'infermatrix-copilot[mcp]'\n") + return 1 + except Exception as exc: # noqa: BLE001 — the reason must be on disk + lifecycle.record("exit", reason="build_error", rc=1, + detail=f"{type(exc).__name__}: {exc}") + raise + rc = 0 + reason, detail = "stdin_closed", "" + try: + server.run() # stdio transport by default; returns at stdin EOF + except KeyboardInterrupt: + reason, detail, rc = "signal", "KeyboardInterrupt", 130 + except SystemExit as exc: + reason, detail = "system_exit", repr(exc.code) + rc = exc.code if isinstance(exc.code, int) else 1 + except BaseException as exc: # noqa: BLE001 — logged, then re-raised + reason, detail, rc = "exception", f"{type(exc).__name__}: {exc}", 1 + lifecycle.record("exit", reason=reason, detail=detail, rc=rc, + calls=lifecycle.calls, parent_alive=_parent_alive(ppid)) + raise + lifecycle.record("exit", reason=reason, detail=detail, rc=rc, + calls=lifecycle.calls, parent_alive=_parent_alive(ppid)) + return rc + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="infermatrix_copilot.tool_bridge", description="Scoped-tool MCP bridge for harness backends (stdio).") parser.add_argument("--spec", required=True, help="bridge spec JSON path") args = parser.parse_args(argv) - try: - server = build_server(Path(args.spec)) - except ImportError: - import sys - - sys.stderr.write("the tool bridge needs the MCP SDK — " - "pip install 'infermatrix-copilot[mcp]'\n") - return 1 - server.run() # stdio transport by default - return 0 + return run_bridge(Path(args.spec)) if __name__ == "__main__": diff --git a/test/test_adapter_knowledge.py b/test/test_adapter_knowledge.py index 65e633f0e..43da27331 100644 --- a/test/test_adapter_knowledge.py +++ b/test/test_adapter_knowledge.py @@ -175,10 +175,19 @@ def test_live_prompt_keeps_bridge_polls_below_client_deadline(prompt_data): assert "under 20 seconds each" in live assert "sleep 30" not in live assert "Cap per-call sleep at 60 s" not in live - assert "`setsid -f sh -c" in live + assert "setsid -f sh -c" in live assert "plain `nohup ... &` is insufficient" in live assert "echo $$ > \"$1\"; shift; exec \"$@\"" in live assert "bare stdin reader" in live + # the detach is explicit AND verifiable: the launch is one tool call, + # the `ps` sid proof must hold, and polling reads the wrapper's receipt + assert "never run\n a verification command in the foreground" in live + assert "ps -o pid,sid,pgid,stat,etimes,cmd" in live + assert "`sid` must\n equal its own pid" in live + assert 'kill -0 "$(cat "$pidfile")"' in live + assert "tests/outcomes/wrap-*.json" in live + assert '`signal_origin: "external"`' in live + assert "reserved for the wrapper's process tree" in live assert "TEST_TIMEOUT_SEC=7200" in live # GOLDEN_KWARGS default budget assert "stops this attempt's background processes" in live diff --git a/test/test_assembly.py b/test/test_assembly.py index 8a3efb81e..263377804 100644 --- a/test/test_assembly.py +++ b/test/test_assembly.py @@ -1506,6 +1506,61 @@ async def fake_rebase_module(module, **kw): assert r.outputs["state_updates"]["module_worker_runner_status"] == "done" +def test_v3_module_rebase_reruns_interrupted_and_blocks_on_interrupt( + v3_agent_env, settings, trace, monkeypatch): + """A SIGINT/SIGTERM leaves a module `interrupted`: on resume that module + is NOT short-circuited (only done/skipped are) and the agent runs again; + and an attempt that itself ends `interrupted` BLOCKS the step so the run + stops instead of advancing to the next sibling.""" + from infermatrix_copilot.engine.registry import StepRegistry + from infermatrix_copilot.engine.step import FailureKind, StepContext + from infermatrix_copilot.engine.steps import rebase_v3, \ + register_builtin_steps + from infermatrix_copilot.rebase_engine import module_rebase as mr + _, _, repo, run_dir = v3_agent_env + monkeypatch.setattr( + rebase_v3, "_tier_client", + lambda ctx: (object(), SimpleNamespace( + model="m", api_key="k", base_url="", source="global"))) + called = [] + outcomes = iter([ + {"status": "done", "exit_code": 0, "debug_attempts": 0, + "turns": 1, "summary": ""}, + {"status": "interrupted", "exit_code": -1, "debug_attempts": 1, + "turns": 4, "interrupted_by": "SIGTERM", "summary": "cut"}, + ]) + + async def fake_rebase_module(module, **kw): + called.append(module) + return next(outcomes) + monkeypatch.setattr(mr, "rebase_module", fake_rebase_module) + sub = Substate(run_dir, "run-int") + sub.update({"modules": {"worker_runner": {"status": "interrupted"}, + "model_executor": {"status": "done"}}}) + registry = register_builtin_steps(StepRegistry()) + handler = registry.get("rebase.v3_module_rebase").handler + + def ctx_for(item): + return StepContext( + settings=settings, params={}, run_dir=run_dir, trace=trace, + item=item, + state={"task_spec": {"repo": "vllm-omni", "params": {}}, + "run_id": "run-int", "repo_path": str(repo)}) + + # resume: the interrupted module runs again, the done one does not + r = asyncio.run(handler(ctx_for("worker_runner"))) + assert r.ok and called == ["worker_runner"] + assert r.outputs["state_updates"]["module_worker_runner_status"] == "done" + r = asyncio.run(handler(ctx_for("model_executor"))) + assert r.ok and "short-circuit" in r.summary and called == ["worker_runner"] + # an attempt interrupted mid-flight blocks the step (resumable) + sub.update({"modules": {"worker_runner": {"status": "interrupted"}}}) + r = asyncio.run(handler(ctx_for("worker_runner"))) + assert not r.ok and r.failure is FailureKind.BLOCKED + assert "SIGTERM" in r.summary and "--resume" in r.summary + assert r.outputs["state_updates"]["module_worker_runner_status"] == "interrupted" + + @pytest.mark.parametrize("contents, expected", [ ("Suggestion...Passed\n- hook id: suggestion\n", None), (("Migration...(no files to check)Skipped\n" diff --git a/test/test_kb_intake_gate.py b/test/test_kb_intake_gate.py index d38796dc1..3bcb469e7 100644 --- a/test/test_kb_intake_gate.py +++ b/test/test_kb_intake_gate.py @@ -118,7 +118,7 @@ def fake_run(cmd, **kwargs): captured["cmd"] = cmd return SimpleNamespace(stdout="", returncode=0) - monkeypatch.setattr(codex.subprocess, "run", fake_run) + monkeypatch.setattr(codex, "run_harness_cli", fake_run) transport = codex.CodexTransport.__new__(codex.CodexTransport) transport.settings = SimpleNamespace(strict_backend_model="", strict_backend_timeout_s=5) monkeypatch.setattr(transport, "require_cli", lambda: "codex", raising=False) diff --git a/test/test_module_pytest_signals.py b/test/test_module_pytest_signals.py new file mode 100644 index 000000000..c5f0f1075 --- /dev/null +++ b/test/test_module_pytest_signals.py @@ -0,0 +1,254 @@ +"""Module test wrapper process contract (run-20261027-193251: four wrapper +runs on `model_executor` ended at ~30 s with child rc -15/-15/-9/-2, +`timed_out=false`, `watchdog_triggered=false` — a GPU reservation guard +outside the process tree killed the test processes, and the receipt could +not say so). + +Under test (rebase_engine/module_pytest.py + testing/runner.py): + * the real wrapper re-sessions itself (sid == pid) and ignores SIGHUP, + so a harness tool's deadline kill of its process group and the loss of + the tool's stdin do not end a detached test; + * SIGTERM/SIGINT to the wrapper stop the test tree and are recorded as + `signal_origin: wrapper_signal`; + * a child killed by a signal nobody in the tree sent is classified + `signal_origin: external` with a diagnosis in receipt AND log; + * in-process callers (`main(argv)`) never re-session the calling process; + * the runner spawns tests with stdin from /dev/null. +""" + +from __future__ import annotations + +import json +import os +import signal +import subprocess +import sys +import time +from pathlib import Path + +import pytest + +from infermatrix_copilot.rebase_engine import module_pytest +from infermatrix_copilot.rebase_engine.module_pytest import ( + RECEIPT_SCHEMA_VERSION, + WrapperProcess, +) +from infermatrix_copilot.testing.runner import TestOutcome as Outcome + +ROOT = Path(__file__).resolve().parents[1] +pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX process model") + + +@pytest.fixture +def wrapper_env(tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + return { + "PATH": f"{Path(sys.executable).parent}{os.pathsep}{os.defpath}", + "PYTHONPATH": str(ROOT / "src"), + "IMX_TARGET_REPO": str(repo), + "IMX_LOG_DIR": str(tmp_path / "logs"), + "IMX_RUN_ID": "run-20200101-000000-abcdef", + "IMX_GPU_MUTEX": "0", + "CUDA_VISIBLE_DEVICES": "", + "TEST_TIMEOUT_SEC": "60", + } + + +def _receipt(env) -> dict: + files = sorted((Path(env["IMX_LOG_DIR"]) / "tests" / "outcomes").glob("*.json")) + assert len(files) == 1, files + return json.loads(files[0].read_text()) + + +def _log(env) -> str: + [log] = list((Path(env["IMX_LOG_DIR"]) / "tests").glob("*.log")) + return log.read_text(encoding="utf-8", errors="replace") + + +def _spawn_wrapper(env, code: str, **popen_kw) -> subprocess.Popen: + return subprocess.Popen( + [sys.executable, "-m", module_pytest.__name__, "python", "-c", code], + env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, **popen_kw) + + +def _wait_for_child(wrapper_pid: int, marker: Path, timeout: float = 20) -> int: + """The test command writes its pid to `marker` once it is running.""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if marker.exists() and marker.read_text().strip(): + return int(marker.read_text().strip()) + time.sleep(0.05) + raise AssertionError("test child never started") + + +def _sleeper(marker: Path) -> str: + return (f"import os, time; open({str(marker)!r}, 'w').write(str(os.getpid()));" + " time.sleep(120)") + + +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + +# -- unit: classification -------------------------------------------------------------- + +@pytest.mark.parametrize("outcome, received, expected", [ + (Outcome(rc=0), [], (None, None)), + (Outcome(rc=1), [], (None, None)), + (Outcome(rc=124, timed_out=True), [], (None, "timeout")), + (Outcome(rc=-15, watchdog_triggered=True), [], ("SIGTERM", "watchdog")), + (Outcome(rc=-15), ["SIGTERM"], ("SIGTERM", "wrapper_signal")), + (Outcome(rc=-15), [], ("SIGTERM", "external")), + (Outcome(rc=-9), [], ("SIGKILL", "external")), + (Outcome(rc=-2), [], ("SIGINT", "external")), +]) +def test_classify_signal_origin(outcome, received, expected): + process = WrapperProcess() + process.received = [{"signal": s, "ts": 0.0} for s in received] + verdict = process.classify(outcome) + assert (verdict["signal"], verdict["signal_origin"]) == expected + if expected[1] == "external": + assert "external supervisor" in verdict["signal_diagnosis"] + assert "not reserved for this process tree" in verdict["signal_diagnosis"] + else: + assert verdict["signal_diagnosis"] is None + + +def test_in_process_main_never_resessions_the_caller(tmp_path, monkeypatch, wrapper_env): + """Tests (and any in-process caller) pass argv: the wrapper must not + `setsid` or install handlers in THEIR process.""" + for k, v in wrapper_env.items(): + monkeypatch.setenv(k, v) + detached = [] + monkeypatch.setattr(WrapperProcess, "detach", + lambda self: detached.append("detach")) + monkeypatch.setattr(WrapperProcess, "install_signal_handlers", + lambda self, runner: detached.append("handlers")) + assert module_pytest.main(["python", "-c", "pass"]) == 0 + assert detached == [] + receipt = _receipt(wrapper_env) + assert receipt["schema_version"] == RECEIPT_SCHEMA_VERSION + assert receipt["wrapper_detached"] is False + assert receipt["signal"] is None and receipt["signal_origin"] is None + assert receipt["wrapper_pid"] == os.getpid() + assert receipt["wrapper_ppid"] == os.getppid() + + +# -- real wrapper process ------------------------------------------------------------- + +def test_wrapper_detaches_and_ignores_sighup(tmp_path, wrapper_env): + marker = tmp_path / "child.pid" + proc = _spawn_wrapper(wrapper_env, _sleeper(marker)) + try: + child = _wait_for_child(proc.pid, marker) + # own session: a tool-shell group kill cannot reach it + assert os.getsid(proc.pid) == proc.pid + assert os.getsid(child) != proc.pid # the test has its own session too + proc.send_signal(signal.SIGHUP) + proc.stdin.close() # the launching tool's pipe goes away + time.sleep(1.0) + assert proc.poll() is None and _alive(child) + # now stop it the sanctioned way + proc.send_signal(signal.SIGTERM) + proc.wait(timeout=30) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + receipt = _receipt(wrapper_env) + assert receipt["wrapper_detached"] is True + assert receipt["wrapper_session_leader"] is True + assert receipt["wrapper_sid"] == receipt["wrapper_pid"] == proc.pid + assert "SIGHUP" in receipt["wrapper_signals_ignored"] + assert receipt["wrapper_signals_received"] == ["SIGTERM"] + + +def test_wrapper_sigterm_stops_tree_and_records_wrapper_signal(tmp_path, wrapper_env): + marker = tmp_path / "child.pid" + proc = _spawn_wrapper(wrapper_env, _sleeper(marker)) + try: + child = _wait_for_child(proc.pid, marker) + proc.send_signal(signal.SIGTERM) + rc = proc.wait(timeout=60) + for _ in range(100): + if not _alive(child): + break + time.sleep(0.1) + assert not _alive(child), "test child outlived the signalled wrapper" + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + stderr = proc.stderr.read().decode() + receipt = _receipt(wrapper_env) + assert receipt["rc"] == -signal.SIGTERM # the child's signal rc + assert rc == (-signal.SIGTERM) & 0xFF # wrapper exit mirrors it + assert receipt["timed_out"] is False and receipt["watchdog_triggered"] is False + assert receipt["signal"] == "SIGTERM" + assert receipt["signal_origin"] == "wrapper_signal" + assert receipt["signal_diagnosis"] is None + assert receipt["wrapper_signals_received"] == ["SIGTERM"] + assert "stopped by SIGTERM" in stderr + assert "signal postmortem" in _log(wrapper_env) + + +def test_external_kill_of_the_test_is_diagnosed(tmp_path, wrapper_env): + """The evidence case: something outside the tree SIGKILLs the test + process. The wrapper survives to write the receipt, which names the + signal, classifies it `external` and carries the diagnosis; the log + gets the same footer for the debug agent.""" + marker = tmp_path / "child.pid" + proc = _spawn_wrapper(wrapper_env, _sleeper(marker)) + try: + child = _wait_for_child(proc.pid, marker) + os.kill(child, signal.SIGKILL) + rc = proc.wait(timeout=60) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + receipt = _receipt(wrapper_env) + assert receipt["rc"] == -signal.SIGKILL and rc == (-signal.SIGKILL) & 0xFF + assert receipt["timed_out"] is False and receipt["watchdog_triggered"] is False + assert receipt["signal"] == "SIGKILL" + assert receipt["signal_origin"] == "external" + assert "external supervisor" in receipt["signal_diagnosis"] + assert receipt["wrapper_signals_received"] == [] + log = _log(wrapper_env) + assert "signal=SIGKILL origin=external" in log + assert "external supervisor" in log + assert "external supervisor" in proc.stderr.read().decode() + # the receipt's log hash covers the footer (written before the receipt) + import hashlib + [log_path] = list((Path(wrapper_env["IMX_LOG_DIR"]) / "tests").glob("*.log")) + assert receipt["log_sha256"] == hashlib.sha256(log_path.read_bytes()).hexdigest() + + +def test_foreground_opt_out_keeps_the_callers_session(tmp_path, wrapper_env): + env = dict(wrapper_env, IMX_WRAPPER_FOREGROUND="1") + proc = _spawn_wrapper(env, "pass", start_new_session=True) + proc.wait(timeout=60) + receipt = _receipt(env) + assert receipt["wrapper_detached"] is False + assert receipt["wrapper_detach_error"] == "IMX_WRAPPER_FOREGROUND=1" + assert "SIGHUP" in receipt["wrapper_signals_ignored"] # still hangup-safe + assert receipt["rc"] == 0 + + +def test_test_child_gets_devnull_stdin(tmp_path, wrapper_env): + """A test must never inherit the wrapper's stdin (the tool pipe).""" + marker = tmp_path / "stdin.txt" + code = (f"import sys; open({str(marker)!r}, 'w').write(" + "repr(sys.stdin.read()) if sys.stdin else 'none')") + proc = _spawn_wrapper(wrapper_env, code) + proc.stdin.write(b"from the tool pipe\n") + proc.stdin.flush() + proc.wait(timeout=60) + assert marker.read_text() == "''" # /dev/null: EOF, not our bytes + assert _receipt(wrapper_env)["rc"] == 0 diff --git a/test/test_run_shutdown.py b/test/test_run_shutdown.py new file mode 100644 index 000000000..ec9da1d94 --- /dev/null +++ b/test/test_run_shutdown.py @@ -0,0 +1,435 @@ +"""Graceful shutdown of a run on SIGINT/SIGTERM (run-20261027-193251: the +engine inherited SIG_IGN for SIGINT from its nohup launcher, so the GPU +scheduler's `--timeout` SIGINT did nothing and the reservation guard then +killed the test processes underneath it one by one). + +Contract under test (engine/lifecycle.py + app/workflow_execution.py + +rebase_engine/module_rebase.py + engine/steps/rebase_v3.py): + * handlers are installed over an inherited SIG_IGN and restored after; + * the shutdown worker terminates registered harness children (whole + tree), runs the module hooks, and exits `128 + signal` at the hard + deadline if the run does not unwind; + * the in-flight module is left `interrupted` (never `running`/`done`) + with a `module_interrupted` trace event, on both the cooperative and + the cancelled path; + * the module step BLOCKS on `interrupted`, the wave gate treats it as + not-done, and the substate short-circuit re-runs it on resume; + * the run is marked `interrupted` in run_status.json with this pid. +""" + +from __future__ import annotations + +import asyncio +import os +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot import run_status as rs +from infermatrix_copilot.engine import lifecycle +from infermatrix_copilot.rebase_engine.substate import Substate +from infermatrix_copilot.run_trace import RunTrace + + +@pytest.fixture(autouse=True) +def _clean_shutdown_state(): + lifecycle._reset_shutdown_state_for_tests() + yield + lifecycle._reset_shutdown_state_for_tests() + + +def _alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + return True + + +# -- handler installation ------------------------------------------------------- + +def test_handlers_override_inherited_sig_ign_and_restore(tmp_path): + """The launcher's `nohup ... &` leaves SIGINT ignored; Python keeps the + inherited disposition, so `install_run_signal_handlers` must install + over SIG_IGN unconditionally — and put it back when the run ends.""" + before_int = signal.signal(signal.SIGINT, signal.SIG_IGN) + before_term = signal.getsignal(signal.SIGTERM) + try: + restore = lifecycle.install_run_signal_handlers(tmp_path, "run-x") + assert callable(signal.getsignal(signal.SIGINT)) + assert signal.getsignal(signal.SIGINT) is not signal.SIG_IGN + assert callable(signal.getsignal(signal.SIGTERM)) + restore() + assert signal.getsignal(signal.SIGINT) is signal.SIG_IGN + assert signal.getsignal(signal.SIGTERM) == before_term + finally: + signal.signal(signal.SIGINT, before_int) + + +def test_handler_records_request_with_run_identity(tmp_path, monkeypatch, capsys): + """Delivering the signal to this process accepts exactly one request + (repeats are counted, not restarted) carrying the run dir and id, and + the exit code follows the shell convention 128 + signal.""" + monkeypatch.setattr(lifecycle, "request_shutdown", + lambda signum, **kw: lifecycle.ShutdownRequest( + signum=signum, **{k: v for k, v in kw.items() + if k in ("run_dir", "run_id")})) + restore = lifecycle.install_run_signal_handlers(tmp_path, "run-sig", + signals=(signal.SIGTERM,)) + try: + handler = signal.getsignal(signal.SIGTERM) + req = handler(signal.SIGTERM, None) + finally: + restore() + assert req.signal_name == "SIGTERM" and req.exit_code == 143 + assert req.run_dir == tmp_path and req.run_id == "run-sig" + assert "resumable: --resume run-sig" in capsys.readouterr().out + + +def test_request_is_idempotent_and_counts_repeats(): + exits: list[int] = [] + first = lifecycle.request_shutdown(signal.SIGINT, hard_deadline_s=0.0, + grace_s=0.1, exit_fn=exits.append) + again = lifecycle.request_shutdown(signal.SIGTERM, hard_deadline_s=0.0, + grace_s=0.1, exit_fn=exits.append) + assert again is first and first.repeats == 1 + assert first.signum == signal.SIGINT and first.exit_code == 130 + assert lifecycle.shutdown_requested() is first + + +# -- the shutdown worker -------------------------------------------------------- + +def test_worker_terminates_registered_child_tree_then_exits(tmp_path): + """A registered harness child (stand-in for cursor-agent) and ITS child + (stand-in for the MCP bridge) are both gone after the request; with + the run task never unwinding, the worker exits with 128 + signal.""" + pidfile = tmp_path / "grandchild.pid" + proc = subprocess.Popen( + ["bash", "--norc", "-c", + f"sleep 300 & echo $! > {pidfile}; wait"], + start_new_session=True) + lifecycle.register_child(proc) + deadline = time.monotonic() + 10 + while not pidfile.exists() or not pidfile.read_text().strip(): + assert time.monotonic() < deadline, "grandchild never started" + time.sleep(0.05) + grandchild = int(pidfile.read_text().strip()) + exited = threading.Event() + codes: list[int] = [] + + def fake_exit(code: int) -> None: + codes.append(code) + exited.set() + + # a bound loop that never unwinds: the hard deadline must fire + loop = asyncio.new_event_loop() + try: + task_holder = {} + + async def park(): + task_holder["t"] = asyncio.current_task() + await asyncio.sleep(3600) + + t = loop.create_task(park()) + loop.run_until_complete(asyncio.sleep(0)) + lifecycle.bind_run_task(loop, t) + lifecycle.request_shutdown(signal.SIGTERM, run_dir=tmp_path, run_id="run-w", + hard_deadline_s=1.0, grace_s=0.5, + exit_fn=fake_exit) + assert exited.wait(timeout=30), "worker never reached exit_fn" + assert codes == [143] + assert proc.wait(timeout=10) < 0 # killed by signal, not exited + for _ in range(50): + if not _alive(grandchild): + break + time.sleep(0.1) + assert not _alive(grandchild), "bridge stand-in survived the shutdown" + # the bound task was asked to cancel + loop.run_until_complete(asyncio.sleep(0)) + assert t.cancelled() or t.cancelling() + finally: + t.cancel() + try: + loop.run_until_complete(asyncio.sleep(0)) + except BaseException: # noqa: BLE001 + pass + loop.close() + + +def test_worker_runs_hooks_and_returns_when_run_unwinds(): + """With no bound run task (the run left the loop), the worker runs the + hooks and returns WITHOUT calling exit — the main thread finishes the + terminal protocol itself.""" + seen: list[str] = [] + exits: list[int] = [] + lifecycle.register_shutdown_hook(lambda req: seen.append(req.signal_name)) + lifecycle.request_shutdown(signal.SIGINT, hard_deadline_s=5.0, grace_s=0.1, + exit_fn=exits.append) + deadline = time.monotonic() + 5 + while not seen and time.monotonic() < deadline: + time.sleep(0.02) + assert seen == ["SIGINT"] + time.sleep(0.3) + assert exits == [] + + +def test_run_guarded_binds_and_unbinds_the_run_task(tmp_path): + """`run_guarded` makes its own task cancellable by the shutdown worker + and forgets it on exit — the worker's "run left the loop" check.""" + seen = {} + + async def body(): + seen["bound"] = lifecycle._run_loop is not None + return "ok" + + assert asyncio.run(lifecycle.run_guarded(body(), tmp_path)) == "ok" + assert seen["bound"] is True + assert lifecycle._run_loop is None + + +def test_run_guarded_cancellation_still_finalizes(tmp_path): + """The forced path: the task is cancelled mid-run; finalizers (GPU lock + release, scratch teardown) still run, then CancelledError propagates.""" + done = [] + lifecycle.register_finalizer(tmp_path, lambda outcome: done.append(outcome)) + + async def body(): + lifecycle.request_shutdown(signal.SIGINT, hard_deadline_s=30, grace_s=0.1, + exit_fn=lambda code: None) + await asyncio.sleep(30) + + with pytest.raises(asyncio.CancelledError): + asyncio.run(lifecycle.run_guarded(body(), tmp_path)) + assert len(done) == 1 and done[0] is None + + +# -- harness transport child registration ----------------------------------------- + +def test_run_harness_cli_registers_child_and_kills_tree_on_timeout(): + from infermatrix_copilot.providers.base import run_harness_cli + + registered: list[int] = [] + orig = lifecycle.register_child + + def spy(proc): + registered.append(proc.pid) + orig(proc) + lifecycle.register_child = spy + try: + with pytest.raises(subprocess.TimeoutExpired): + run_harness_cli( + ["bash", "--norc", "-c", "sleep 300 & wait"], + input="", cwd=None, env=None, timeout=0.5) + finally: + lifecycle.register_child = orig + assert len(registered) == 1 + assert lifecycle.registered_children() == [] # unregistered after + + +def test_run_harness_cli_returns_completed_process_and_unregisters(): + from infermatrix_copilot.providers.base import run_harness_cli + cp = run_harness_cli([sys.executable, "-c", + "import sys; print(sys.stdin.read().upper())"], + input="hello", cwd=None, env=None, timeout=30) + assert cp.returncode == 0 and cp.stdout.strip() == "HELLO" + assert lifecycle.registered_children() == [] + + +# -- run status + workflow terminal protocol ---------------------------------------- + +def test_mark_interrupted_stamps_this_pid_and_run_id(tmp_path): + run_dir = tmp_path / "run-20261027-000000-abcdef" + run_dir.mkdir() + rec = rs.mark_interrupted(run_dir, note="SIGINT") + assert rec["state"] == rs.INTERRUPTED + assert rec["child_pid"] == os.getpid() # not a never-launched reservation + assert rec["run_id"] == run_dir.name + assert rec["note"] == "SIGINT" + assert rs.read_status(run_dir)["state"] == rs.INTERRUPTED + + +def test_workflow_interrupted_protocol(tmp_path, capsys): + from infermatrix_copilot.app.workflow_execution import WorkflowExecution + run_dir = tmp_path / "run-20261027-000000-abcdef" + run_dir.mkdir() + trace = RunTrace(run_dir / "run_trace.jsonl") + req = lifecycle.ShutdownRequest(signum=signal.SIGINT, run_dir=run_dir, + run_id=run_dir.name) + outcome = SimpleNamespace(step_results={ + "a": SimpleNamespace(ok=True, summary="done"), + "b": SimpleNamespace(ok=False, summary="interrupted")}) + result = WorkflowExecution._interrupted(run_dir, trace, req, outcome) + assert result.exit_code == 130 + assert "resumable" in result.blocked_reason + assert rs.read_status(run_dir)["state"] == rs.INTERRUPTED + [ev] = trace.events("run_interrupted") + assert ev["signal"] == "SIGINT" and ev["exit_code"] == 130 + assert ev["steps_completed"] == ["a"] + out = capsys.readouterr().out + assert "interrupted (SIGINT)" in out + + +# -- module loop: interrupted substate -------------------------------------------------- + +def _module_setup(tmp_path, monkeypatch, attempts): + """`rebase_module` with the harness attempt replaced: `attempts` is a + list of coroutine factories consumed in order.""" + from infermatrix_copilot.rebase_engine import module_rebase as mr + from infermatrix_copilot.rebase_engine.prompt_builder import ModulePromptData + + run_dir = tmp_path / "run" + (run_dir / "signals").mkdir(parents=True) + templates = tmp_path / "templates" + templates.mkdir() + (templates / "module.tmpl").write_text("module {MODULE_NAME}\n") + (templates / "debug.tmpl").write_text("debug {module}: {traceback}\n") + data = ModulePromptData( + templates_dir=templates, template="module.tmpl", + pytest_wrapper="pytest", template_live="module.tmpl", + debug_prompt_template="debug {module}: {traceback}", + module_vllm_paths={"mod": "vllm/x"}, module_omni_files={"mod": "omni/x"}, + module_test_map={"mod": ["tests/x"]}) + config = mr.ModuleRunConfig( + vllm_path=str(tmp_path), omni_path=str(tmp_path), script_dir=str(tmp_path), + model="m", log_dir=str(run_dir), signal_dir=str(run_dir / "signals"), + backend="cursor", max_debug_retries=3) + queue = list(attempts) + + async def fake_attempt(prompt, **kw): + return await queue.pop(0)() + monkeypatch.setattr(mr, "_harness_attempt", fake_attempt) + monkeypatch.setattr(mr, "build_module_prompt", lambda *a, **kw: "prompt") + monkeypatch.setattr(mr, "build_debug_prompt", lambda *a, **kw: "debug") + sub = Substate(run_dir, "run-int") + trace = RunTrace(run_dir / "run_trace.jsonl") + + def run(): + return asyncio.run(mr.rebase_module( + "mod", client=None, config=config, prompt_data=data, tool_defs=[], + extra_tools={}, substate=sub, trace=trace)) + return run, sub, trace + + +def test_module_cooperative_interrupt_marks_interrupted_without_debug_retry( + tmp_path, monkeypatch): + """The attempt fails while a shutdown is pending: no debug retry, the + module is `interrupted` (not failed/running) with the trace event.""" + async def failing(): + lifecycle.request_shutdown(signal.SIGTERM, hard_deadline_s=30, grace_s=0.1, + exit_fn=lambda code: None) + return {"done": False, "text": "cut", "turns": 3, "plan_done": False} + + async def never(): + raise AssertionError("debug retry ran during shutdown") + + run, sub, trace = _module_setup(tmp_path, monkeypatch, [failing, never]) + outcome = run() + assert outcome["status"] == "interrupted" + assert outcome["interrupted_by"] == "SIGTERM" + assert sub.read()["modules"]["mod"]["status"] == "interrupted" + [ev] = trace.events("module_interrupted") + assert ev["signal"] == "SIGTERM" and ev["step"] == "rebase.module.mod" + assert "interrupted by SIGTERM" in (tmp_path / "run" / "agents" / + "module-mod.log").read_text() + + +def test_module_completed_before_signal_stays_done(tmp_path, monkeypatch): + """A module whose attempt finished `done` just as the signal arrived + keeps its result — interrupting must not discard verified work.""" + async def done(): + lifecycle.request_shutdown(signal.SIGINT, hard_deadline_s=30, grace_s=0.1, + exit_fn=lambda code: None) + return {"done": True, "text": "ok", "turns": 2, "plan_done": True} + + # cooperative path only: the worker thread's hook would otherwise race + # the attempt's return (the hook guards on in-flight; see the hook test) + monkeypatch.setattr(lifecycle, "_shutdown_worker", lambda *a, **kw: None) + run, sub, trace = _module_setup(tmp_path, monkeypatch, [done]) + assert run()["status"] == "done" + assert sub.read()["modules"]["mod"]["status"] == "done" + assert list(trace.events("module_interrupted")) == [] + + +def test_module_cancelled_mid_attempt_marks_interrupted_and_reraises( + tmp_path, monkeypatch): + """The forced path: the run task is cancelled while the harness attempt + is in flight; the module is marked and CancelledError propagates.""" + async def hanging(): + lifecycle.request_shutdown(signal.SIGINT, hard_deadline_s=30, grace_s=0.1, + exit_fn=lambda code: None) + raise asyncio.CancelledError() + + run, sub, trace = _module_setup(tmp_path, monkeypatch, [hanging]) + with pytest.raises(asyncio.CancelledError): + run() + assert sub.read()["modules"]["mod"]["status"] == "interrupted" + assert sub.read()["modules"]["mod"]["interrupted_by"] == "SIGINT" + assert len(list(trace.events("module_interrupted"))) == 1 + assert lifecycle._shutdown_hooks == [] # hook unregistered on exit + + +def test_module_shutdown_hook_marks_in_flight_module_once(tmp_path, monkeypatch): + """While an attempt is in flight the module loop has a hook registered; + the shutdown worker calling it leaves the same durable mark, and a + second call (hook + cooperative path) does not duplicate the event.""" + captured = {} + + async def in_flight(): + [hook] = list(lifecycle._shutdown_hooks) + req = lifecycle.ShutdownRequest(signum=signal.SIGTERM) + hook(req) + hook(req) + captured["during"] = Substate(tmp_path / "run", "run-int").read() + lifecycle.request_shutdown(signal.SIGTERM, hard_deadline_s=30, grace_s=0.1, + exit_fn=lambda code: None) + return {"done": False, "text": "", "turns": 0, "plan_done": False} + + monkeypatch.setattr(lifecycle, "_shutdown_worker", lambda *a, **kw: None) + run, sub, trace = _module_setup(tmp_path, monkeypatch, [in_flight]) + assert run()["status"] == "interrupted" + assert captured["during"]["modules"]["mod"]["status"] == "interrupted" + assert len(list(trace.events("module_interrupted"))) == 1 + + +def test_module_shutdown_hook_ignores_an_attempt_that_already_returned( + tmp_path, monkeypatch): + """After the attempt returned, the hook defers to the cooperative path: + a late worker thread must not flip a finished module.""" + hooks = {} + + async def finished(): + hooks["hook"] = list(lifecycle._shutdown_hooks)[0] + return {"done": True, "text": "ok", "turns": 1, "plan_done": True} + + monkeypatch.setattr(lifecycle, "_shutdown_worker", lambda *a, **kw: None) + run, sub, trace = _module_setup(tmp_path, monkeypatch, [finished]) + assert run()["status"] == "done" + hooks["hook"](lifecycle.ShutdownRequest(signum=signal.SIGTERM)) + assert sub.read()["modules"]["mod"]["status"] == "done" + assert list(trace.events("module_interrupted")) == [] + + +# -- step + wave gate + resume -------------------------------------------------------- + +def test_wave_gate_treats_interrupted_as_not_done(tmp_path, settings, trace): + from infermatrix_copilot.engine.registry import StepRegistry + from infermatrix_copilot.engine.step import StepContext + from infermatrix_copilot.engine.steps import register_builtin_steps + run_dir = tmp_path / "run" + run_dir.mkdir() + Substate(run_dir, "run-g").update({"modules": { + "a": {"status": "done"}, "b": {"status": "interrupted"}}}) + registry = register_builtin_steps(StepRegistry()) + ctx = StepContext(settings=settings, params={}, run_dir=run_dir, trace=trace, + state={"task_spec": {"repo": "vllm-omni", "params": {}}, + "run_id": "run-g", "wave1_modules": ["a", "b"], + "wave2_modules": ["c"]}) + r = asyncio.run(registry.get("rebase.v3_wave_gate").handler(ctx)) + assert r.ok and r.outputs["state_updates"]["wave2_modules"] == [] + assert "b" in r.summary diff --git a/test/test_tool_bridge_lifecycle.py b/test/test_tool_bridge_lifecycle.py new file mode 100644 index 000000000..d65cf123b --- /dev/null +++ b/test/test_tool_bridge_lifecycle.py @@ -0,0 +1,313 @@ +"""Tool-bridge liveness (run-20261027-193251: `infermatrix-tools` discovery +failed mid-session after a 30-minute synchronous plan review froze the +server's event loop; the harness then treated the server as gone for the +remaining hours, with no record of whether the bridge process had died). + +Under test: + * every registered handler is OFFLOADED to a worker thread, so a slow + tool does not block another call or a ping (`_offload`); + * the server process ignores SIGINT/SIGHUP (a stdio server lives as long + as its stdin) and exits when stdin closes, writing start/exit (with a + reason) to `bridge/.lifecycle.jsonl`; + * slow calls are logged (`slow_tool`), which is the only client-side + evidence of a deadline cut; + * the module loop classifies the lifecycle log (`bridge_health`) and + records `bridge_unavailable` with the bridge's exit reason. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path + +import pytest + +from infermatrix_copilot.rebase_engine.module_rebase import bridge_health +from infermatrix_copilot.scopes import read_only_scope +from infermatrix_copilot.tool_bridge import ( + BridgeLifecycle, + _offload, + build_server, + lifecycle_log_path, + read_lifecycle_events, + timed_call, + write_bridge_spec, +) + +ROOT = Path(__file__).resolve().parents[1] + + +def _spec(tmp_path: Path) -> tuple[Path, Path, Path]: + worktree = tmp_path / "worktree" + worktree.mkdir() + (worktree / "a.py").write_text("x = 1\n", encoding="utf-8") + run_dir = tmp_path / "run" + run_dir.mkdir() + scope = read_only_scope() + scope = type(scope)(name=scope.name, allowed_tools=scope.allowed_tools, + read_only=True, root=str(worktree)) + spec = write_bridge_spec(run_dir=run_dir, step_name="rebase.module.x", + scope=scope, repo="demo") + return spec, worktree, run_dir + + +# -- offloading ----------------------------------------------------------------- + +def test_offload_keeps_signature_and_runs_off_loop(): + import inspect + + def handler(path: str, offset: int = 0) -> str: + return f"{threading.current_thread().name}:{path}:{offset}" + + wrapped = _offload(handler) + assert inspect.iscoroutinefunction(wrapped) + assert list(inspect.signature(wrapped).parameters) == ["path", "offset"] + out = asyncio.run(wrapped("p", offset=3)) + assert out.endswith(":p:3") + assert not out.startswith("MainThread") # ran on a worker thread + + +def test_slow_tool_does_not_block_a_concurrent_call(tmp_path, monkeypatch): + """The regression, through the real FastMCP server: with inline sync + handlers the second call waits for the first; offloaded, `list_dir` + returns while `read_file` is still blocked in its handler.""" + import infermatrix_copilot.tool_bridge as tb + spec, _worktree, _run_dir = _spec(tmp_path) + release = threading.Event() + orig_make = tb.make_dispatcher + + def gated_make(*args, **kwargs): + inner = orig_make(*args, **kwargs) + + def call(name, arguments): + if name == "read_file": + assert release.wait(timeout=10), "slow call never released" + return inner(name, arguments) + return call + monkeypatch.setattr(tb, "make_dispatcher", gated_make) + server = build_server(spec) + + async def scenario(): + slow = asyncio.ensure_future(server.call_tool("read_file", {"path": "a.py"})) + fast = await asyncio.wait_for( + server.call_tool("list_dir", {"path": "."}), timeout=10) + assert not slow.done() # still blocked — and we got served + release.set() + return fast, await asyncio.wait_for(slow, timeout=10) + + fast, slow = asyncio.run(scenario()) + assert "a.py" in str(fast) and "x = 1" in str(slow) + + +def test_offloaded_handlers_run_concurrently(tmp_path, monkeypatch): + """Direct proof of concurrency: two offloaded calls into one blocking + handler overlap in time (both are inside before either leaves).""" + inside = threading.Barrier(2, timeout=10) + + def blocking(path: str) -> str: + inside.wait() # both calls must be inside simultaneously + return path + + wrapped = _offload(blocking) + + async def both(): + return await asyncio.gather(wrapped("a"), wrapped("b")) + + assert asyncio.run(both()) == ["a", "b"] + + +# -- lifecycle log ---------------------------------------------------------------- + +def test_timed_call_logs_only_slow_calls(tmp_path): + lifecycle = BridgeLifecycle(tmp_path / "x.lifecycle.jsonl") + calls = [] + + def call(name, args): + calls.append(name) + time.sleep(0.05 if name == "slow" else 0) + return "ok" + + bound = timed_call(call, lifecycle, slow_after_s=0.02) + assert bound("fast", {}) == "ok" and bound("slow", {}) == "ok" + assert lifecycle.calls == 2 + events = read_lifecycle_events(tmp_path / "x.json") + assert [e["kind"] for e in events] == ["slow_tool"] + assert events[0]["tool"] == "slow" and events[0]["elapsed_s"] >= 0.05 + assert timed_call(call, None) is call + + +def test_lifecycle_write_failure_never_raises(tmp_path): + lifecycle = BridgeLifecycle(tmp_path / "a.py" / "impossible.jsonl") + (tmp_path / "a.py").write_text("") # parent is a file: mkdir fails + lifecycle.record("start") # must not raise + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX signals") +def test_bridge_process_survives_sigint_sighup_and_exits_on_stdin_close(tmp_path): + """End-to-end on the real stdio server: SIGINT and SIGHUP are ignored + (the process stays alive), closing stdin ends it with rc 0, and the + lifecycle log carries start + exit(reason=stdin_closed).""" + pytest.importorskip("mcp") + spec, _worktree, run_dir = _spec(tmp_path) + env = {"PATH": os.environ.get("PATH", ""), "PYTHONPATH": str(ROOT / "src"), + "HOME": str(tmp_path)} + proc = subprocess.Popen( + [sys.executable, "-m", "infermatrix_copilot.tool_bridge", "--spec", str(spec)], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + env=env, start_new_session=True) + try: + log = lifecycle_log_path(spec) + deadline = time.monotonic() + 30 + while time.monotonic() < deadline: + if any(e["kind"] == "start" for e in read_lifecycle_events(spec)): + break + time.sleep(0.1) + else: + pytest.fail(f"bridge never logged start: {proc.stderr.read()!r}") + # dispositions are installed before the start event is written, so + # the signals may go out as soon as it appears + proc.send_signal(signal.SIGINT) + proc.send_signal(signal.SIGHUP) + time.sleep(1.0) + assert proc.poll() is None, f"bridge died on SIGINT/SIGHUP: rc={proc.returncode}" + proc.stdin.close() + rc = proc.wait(timeout=30) + assert rc == 0, proc.stderr.read() + events = read_lifecycle_events(spec) + kinds = [e["kind"] for e in events] + assert kinds[0] == "start" and kinds[-1] == "exit" + assert events[0]["pid"] == proc.pid and events[0]["spec"] == str(spec) + assert events[-1]["reason"] == "stdin_closed" + assert events[-1]["parent_alive"] is True # we, the launcher, are alive + assert log.parent == run_dir / "bridge" + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + + +# -- module-loop classification --------------------------------------------------------- + +def _write_events(spec: Path, events: list[dict]) -> None: + path = lifecycle_log_path(spec) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("".join(json.dumps(e) + "\n" for e in events)) + + +@pytest.mark.parametrize("events, verdict, detail_has", [ + ([], "never_started", ""), + ([{"kind": "start"}], "no_exit_recorded", ""), + ([{"kind": "start"}, {"kind": "exit", "reason": "stdin_closed", + "parent_alive": False, "rc": 0}], "ok", ""), + ([{"kind": "start"}, {"kind": "exit", "reason": "exception", "rc": 1, + "detail": "BrokenPipeError: x", "parent_alive": True}], + "exited_early", "BrokenPipeError"), + ([{"kind": "start"}, {"kind": "exit", "reason": "stdin_closed", + "parent_alive": True, "rc": 0}], "exited_early", "stdin_closed"), + ([{"kind": "start"}, {"kind": "exit", "reason": "stdin_closed", "rc": 0}, + {"kind": "start"}], "restarted", ""), +]) +def test_bridge_health_verdicts(tmp_path, events, verdict, detail_has): + spec = tmp_path / "bridge" / "s.json" + now = time.time() + _write_events(spec, [{"ts": now + i, **e} for i, e in enumerate(events)]) + report = bridge_health(spec, since_ts=now - 5, settle_s=0.0) + assert report["verdict"] == verdict + assert detail_has in report["detail"] + assert report["starts"] == sum(e["kind"] == "start" for e in events) + + +def test_bridge_health_ignores_events_from_an_earlier_session(tmp_path): + spec = tmp_path / "bridge" / "s.json" + now = time.time() + _write_events(spec, [{"ts": now - 100, "kind": "start"}, + {"ts": now - 90, "kind": "exit", "reason": "signal"}]) + assert bridge_health(spec, since_ts=now - 5, settle_s=0.0)["verdict"] == "never_started" + + +def test_bridge_health_counts_slow_tools(tmp_path): + spec = tmp_path / "bridge" / "s.json" + now = time.time() + _write_events(spec, [ + {"ts": now, "kind": "start"}, + {"ts": now + 1, "kind": "slow_tool", "tool": "run_shell", "elapsed_s": 70}, + {"ts": now + 2, "kind": "slow_tool", "tool": "request_plan_review", + "elapsed_s": 1800}, + {"ts": now + 3, "kind": "exit", "reason": "stdin_closed", "parent_alive": False}]) + report = bridge_health(spec, since_ts=now - 1, settle_s=0.0) + assert report["verdict"] == "ok" + assert report["slow_tools"] == 2 + assert report["slowest_tool"] == "request_plan_review" + + +def test_bridge_health_waits_briefly_for_the_exit_event(tmp_path): + """The harness returns before its bridge child finishes: a start with + no exit yet is given `settle_s` to land before being judged.""" + spec = tmp_path / "bridge" / "s.json" + now = time.time() + _write_events(spec, [{"ts": now, "kind": "start"}]) + + def late_exit(): + time.sleep(0.3) + with lifecycle_log_path(spec).open("a") as f: + f.write(json.dumps({"ts": time.time(), "kind": "exit", + "reason": "stdin_closed", "parent_alive": False}) + "\n") + + threading.Thread(target=late_exit, daemon=True).start() + report = bridge_health(spec, since_ts=now - 1, settle_s=3.0) + assert report["verdict"] == "ok" + + +def test_harness_attempt_records_bridge_unavailable(tmp_path, monkeypatch): + """Through `_harness_attempt`: a bridge that exited with an exception + while the harness ran becomes a `bridge_unavailable` trace event with + the exit reason and a note in the result text.""" + from infermatrix_copilot.agent_loop import AgentOutcome + from infermatrix_copilot.providers import registry + from infermatrix_copilot.rebase_engine import module_rebase as mr + from infermatrix_copilot.run_trace import RunTrace + from infermatrix_copilot.scopes import PathScope, ToolScope + + repo = tmp_path / "repo" + run = tmp_path / "run" + (run / "signals").mkdir(parents=True) + repo.mkdir() + trace = RunTrace(run / "bridge_trace.jsonl") + scope = ToolScope(name="rebase-module:fixture", allowed_tools=frozenset({"write_file"}), + path_scope=PathScope(writable=(f"{run}/*",), primary=(f"{run}/*",)), + read_only=False, root=str(repo)) + + class Session: + def run_session(self, req): + # the bridge the harness launched: started, then died mid-session + now = time.time() + _write_events(Path(req.bridge_spec_path), [ + {"ts": now, "kind": "start", "pid": 4242}, + {"ts": now + 1, "kind": "exit", "reason": "exception", "rc": 1, + "detail": "RuntimeError: boom", "parent_alive": True, "pid": 4242}]) + (run / "signals" / "module.fixture.done").write_text("MODULE_DONE fixture\n") + return AgentOutcome(text="finished", iterations=1, tool_calls=1, refusals=[]) + + monkeypatch.setattr(registry, "transport_for_id", lambda *a, **kw: Session()) + monkeypatch.setattr(mr, "BRIDGE_EXIT_SETTLE_S", 0.0) + config = mr.ModuleRunConfig( + vllm_path=str(repo), omni_path=str(repo), script_dir=str(repo), + model="fixture-model", log_dir=str(run), signal_dir=str(run / "signals"), + backend="zcode", paths_spec={"omni_path": str(repo), "vllm_path": str(repo)}) + result = asyncio.run(mr._harness_attempt( + "fixture", module="fixture", config=config, scope=scope, trace=trace, + tool_defs=[], plan_prefix=str(run / "plans" / "module-fixture"), + require_plan_review=False)) + [ev] = list(trace.events("bridge_unavailable")) + assert ev["verdict"] == "exited_early" + assert "RuntimeError: boom" in ev["detail"] + assert ev["step"] == "rebase.module.fixture" + assert "events" not in ev + assert "Tool bridge exited early" in result["text"]