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 4338eccb..51f52be5 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 9081e9dc..f7417453 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 c8f17b1a..e9a9f21b 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 82df37b7..9892e88a 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 795900a5..2e19a534 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 de9d5e3b..27a27ba0 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 abc13f3f..8a3604fd 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 ee56ed54..bb62559e 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 8f402ad8..a1dace53 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 20218677..90631c5a 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 c423bdc8..0e668c3a 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 26b2cf92..b69e6e4c 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 fa27d31b..e35ea229 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 d84413eb..a4423df1 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 98a3f47d..4d71f742 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 d38198da..0650d35a 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 52cf8ef8..7afdd85e 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 c8a362af..b1ebe31e 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 8055fcf6..fb03bd5b 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 0b645a14..cbaa1d27 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 c72f8c40..393e56e8 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 f94fa3c0..2c1bfcb4 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 de6e0deb..1389e56f 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 4d568b90..00b9fd99 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 b1d764e4..5098070f 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 eb6cb7b5..01894d98 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 3eb51d83..ba0d50db 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 65e633f0..43da2733 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 8a3efb81..26337780 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 d38796dc..3bcb469e 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 00000000..c5f0f107 --- /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 00000000..ec9da1d9 --- /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 00000000..d65cf123 --- /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"]