Fix local rebase runtime isolation and API drift validation

adapters/vllm_omni/rebase/api_drift_guard.py

@@ -56,6 +56,9 @@ def find_spec(fullname, path, target=None):
     ("vllm.v1.worker.gpu_model_runner", "GPUModelRunner", "vllm_omni.worker.gpu_ar_model_runner", "GPUARModelRunner"),
     ("vllm.v1.core.sched.scheduler", "Scheduler", "vllm_omni.core.sched.omni_ar_scheduler", "OmniARScheduler"),
     ("vllm.v1.core.sched.scheduler", "Scheduler", "vllm_omni.core.sched.omni_generation_scheduler", "OmniGenerationScheduler"),
+    ("vllm.v1.engine.output_processor", "RequestState", "vllm_omni.outputs.output_processor", "OmniRequestState"),
+    ("vllm.v1.engine.output_processor", "OutputProcessor", "vllm_omni.outputs.output_processor", "MultimodalOutputProcessor"),
+    ("vllm.profiler.wrapper", "WorkerProfiler", "vllm_omni.profiler.omni_torch_profiler", "OmniTorchProfilerWrapper"),
 ]
 
 
@@ -276,9 +279,17 @@ def _check_inheritance() -> list[str]:
                 ):
                     continue
                 missing.append(pname)
-            if missing:
+            positional_kinds = (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
+            base_positional = [p.name for p in base_sig.parameters.values() if p.kind in positional_kinds]
+            omni_positional = [p.name for p in omni_sig.parameters.values() if p.kind in positional_kinds]
+            omni_has_var_positional = any(
+                p.kind == inspect.Parameter.VAR_POSITIONAL for p in omni_sig.parameters.values()
+            )
+            lost_positional = [] if omni_has_var_positional else base_positional[len(omni_positional):]
+            if missing or lost_positional:
                 msg = (
-                    f"MISMATCH {omni_cls_name}.{name}(): missing params {missing}\n"
+                    f"MISMATCH {omni_cls_name}.{name}(): missing params {missing}; "
+                    f"incompatible positional params {lost_positional}\n"
                     f"  Base : {base_sig}\n"
                     f"  Omni : {omni_sig}"
                 )

doc/architecture/SPEC/cli.md

@@ -1,6 +1,6 @@
 # cli/ —— 规范
 
-<!-- verified-against: 2026-10-29 -->
+<!-- verified-against: 2026-10-30 -->
 
 `LOC ~1420（6 个文件） · 接口 + 编排门面 · refactor-status: ok`
 

doc/architecture/SPEC/engine/steps/rebase_v3.md

@@ -1,6 +1,6 @@
 # engine/steps/rebase_v3.py —— 规范
 
-<!-- verified-against: 2026-10-29 -->
+<!-- verified-against: 2026-10-30 -->
 
 `LOC ~2204 · step 库（v3 rebase 装配层） · refactor-status: oversized`
 
@@ -132,7 +132,9 @@ harness 会话通过 MCP tool bridge 拿到同一套工具面，plan gate 在派
   `upstream_commit` 重新确保 scratch。
 - baseline venv（`ensure_baseline_venv`）：硬链接克隆 campaign venv 并换入
   baseline commit 的 wheel，让"omni main 对上次对齐的 wheel 是否也失败"可问；
-  任何失败返回 None，保留保守的回归判定。
+  使用发布 wheel（若配置 URL）或非 editable 安装，删除临时源码后再验证导入；
+  克隆的 console/activation 脚本通过原子替换改为 baseline 路径，不能写穿硬链接
+  修改 campaign。缓存复用也必须通过导入检查。任何失败返回 None，保留保守的回归判定。
 
 ## 操作者重跑与本地超时预算（2026-10-29）
 
@@ -142,9 +144,14 @@ harness 会话通过 MCP tool bridge 拿到同一套工具面，plan gate 在派
   `run_test_loop`：记录过的 timeout/watchdog infra 再跑一次，已通过的 job 不重跑。
   普通 resume 下 infra 仍然粘滞。
 - `rebase.testing.local_timeout_multiplier`（默认 1，须为 >= 1 的数，否则 BLOCKED）
-  只放大本地 `TestJob.timeout_sec`；构建出的 manifest 保留 CI 的
+  放大本地 `TestJob.timeout_sec` 和命令行首的直接 `timeout DURATION`；构建出的 manifest 保留 CI 的
   `timeout_in_minutes`（shell golden 不变）。vllm-omni 取 3：Ready 的 20 分钟 CPU
   Simple step 在共享 rebase 主机上通过 53–88% 用例、零失败时就被杀。
+- 本地已配置虚拟环境时，将行首 `uv pip install --system` 改为
+  `uv pip install --python "$VIRTUAL_ENV/bin/python"`。候选与基线各自使用
+  `_target_test_env` 选定的虚拟环境；不修改 CI manifest，也不向系统 Python 安装依赖。
+- API drift guard 覆盖输出处理器、请求状态及 profiler 基类；除参数名外还检查位置参数容量，
+  防止将上游可按位置传递的可选参数改成仅关键字参数后仍误报兼容。
 
 ## 信号中断的模块：BLOCKED 收口、resume 重跑（2026-10-28）
 

doc/architecture/SPEC/rebase_engine.md

@@ -1,6 +1,6 @@
 # rebase_engine/ —— 规范
 
-<!-- verified-against: 2026-10-29 -->
+<!-- verified-against: 2026-10-30 -->
 
 `LOC ~7500（26 个模块） · repo-rebase-v3 的原生 rebase 引擎 · refactor-status: ok`
 
@@ -88,6 +88,9 @@ runner/LLM/CI client 全部可注入 —— 每个模块都能离线测试。
   取锁（父级 parity）。
 - wheel：每个 uv 调用 `--python` 显式指向**声明的目标 venv** —— 启动
   shell 的 venv 状态绝不影响装到哪；只有 import 验证失败才重试。
+- baseline venv：克隆后的 launcher/activation 路径必须独立且重复修复幂等；
+  baseline 包优先安装冻结 wheel，否则非 editable 安装源码。临时 worktree
+  删除后须再次从 venv 目录验证 import；缓存也须验证实际 import，不能只信版本元数据。
 - CI 创建受守卫：durable op 记录先于 API 调用，恢复按 op id 精确匹配，
   **不确定时绝不重复创建**。
 - 每个被 monitor 的 CI round 持久化实际 monitor 时长，供 run metrics 从 durable

src/infermatrix_copilot/engine/steps/rebase_v3.py

@@ -102,6 +102,29 @@ def _local_timeout_multiplier(testing: Mapping) -> float | StepResult:
     return factor
 
 
+def _local_test_job(job, *, timeout_multiplier: float, virtualenv: bool):
+    """Adapt container commands to the selected local runtime and budget.
+
+    VIRTUAL_ENV is supplied by _target_test_env for both candidate and
+    baseline runs. The CI manifest itself remains unchanged.
+    """
+    command = job.command
+    if virtualenv:
+        command = re.sub(
+            r'(?m)^([ \t]*uv[ \t]+pip[ \t]+install[ \t]+)--system(?=[ \t]|$)',
+            r'\1--python "$VIRTUAL_ENV/bin/python"', command)
+    if timeout_multiplier != 1:
+        # Ready jobs also put direct timeout commands around individual
+        # pytest invocations. Scaling only the outer runner leaves those
+        # inner CI budgets unchanged and still kills valid local runs.
+        command = re.sub(
+            r'(?m)^([ \t]*timeout[ \t]+)(\d+(?:\.\d+)?)([smhd]?)(?=[ \t])',
+            lambda m: (f'{m[1]}{float(m[2]) * timeout_multiplier:g}{m[3]}'),
+            command)
+    return _dc_replace(job, command=command,
+                       timeout_sec=job.timeout_sec * timeout_multiplier)
+
+
 def _tier_client(ctx: StepContext):
     """The run's agent backend via `Settings.tier_target` — the ONLY place a
     tier's model may pair with an endpoint and credential. Returns
@@ -1546,9 +1569,10 @@ def notify_download(key: str, repo_id: str) -> None:
 
     def _job(slug: str) -> TestJob:
         job = manifest_job_to_test_job(jobs_by_slug[slug])
-        if timeout_multiplier != 1:
-            job = _dc_replace(job, timeout_sec=job.timeout_sec * timeout_multiplier)
-        return job
+        return _local_test_job(
+            job, timeout_multiplier=timeout_multiplier,
+            virtualenv=bool(_target_venv(
+                manifest, extra=ctx.settings.expansion_env())))
 
     def _to_result(outcome) -> tl.TestRunResult:
         infra = "timeout" if outcome.timed_out else \

src/infermatrix_copilot/rebase_engine/wheel.py

@@ -749,6 +749,31 @@ def import_ok() -> bool:
 
 # -- baseline runtime ----------------------------------------------------------
 
+def _relocate_venv_scripts(src: Path, dst: Path) -> None:
+    """Point cloned launchers/activation scripts at the clone, without writing
+    through hardlinks into the campaign runtime."""
+    import stat
+
+    source = os.fsencode(str(src))
+    destination = os.fsencode(str(dst))
+    # The baseline directory name starts with the campaign name. Match the
+    # complete path so a healthy cloned launcher is not relocated twice.
+    source_path = re.compile(re.escape(source) + rb"(?=[/\x22\x27\s]|$)")
+    for script in (dst / "bin").iterdir():
+        if script.is_symlink() or not script.is_file():
+            continue
+        data = script.read_bytes()
+        if b"\0" in data or not source_path.search(data):
+            continue
+        replacement = script.with_name(f".{script.name}.relocate-{os.getpid()}")
+        try:
+            replacement.write_bytes(source_path.sub(lambda _: destination, data))
+            replacement.chmod(stat.S_IMODE(script.stat().st_mode))
+            os.replace(replacement, script)
+        finally:
+            replacement.unlink(missing_ok=True)
+
+
 def _clone_tree_hardlinked(src: Path, dst: Path) -> bool:
     """Hardlink clone (near-zero disk, same filesystem). uv replaces files
     by rename, which breaks the links — the campaign venv is never modified
@@ -757,6 +782,7 @@ def _clone_tree_hardlinked(src: Path, dst: Path) -> bool:
     if rc != 0:
         _log(f"baseline-venv clone failed: {err.strip()}")
         return False
+    _relocate_venv_scripts(src, dst)
     return True
 
 
@@ -778,7 +804,8 @@ def ensure_baseline_venv(
     against the wheel main was LAST aligned to"; this builds exactly that
     runtime: hardlink-clone the campaign venv (identical deps), swap in
     the baseline wheel from a throwaway worktree of the upstream at the
-    baseline commit. Returns None on any failure — a missing baseline
+    baseline commit. Installs are non-editable so cleanup cannot remove code
+    the runtime needs; console scripts are relocated to the clone. Returns None on any failure — a missing baseline
     runtime keeps the loop's regression-preserving default and must never
     read as pre-existing."""
     import shutil
@@ -791,9 +818,16 @@ def ensure_baseline_venv(
     python = str(bv / "bin" / "python")
     if (bv / "pyvenv.cfg").is_file():
         installed = read_installed_version(spec, python=python, run=run)
-        if installed and version_matches_commit(installed, baseline_commit):
-            log(f"baseline venv already healthy at {bv}")
-            return bv
+        matches = (installed_wheel_matches(
+            spec, baseline_commit, installed, python=python, run=run)
+            if spec.install_url_template else
+            bool(installed and version_matches_commit(installed, baseline_commit)))
+        if matches:
+            rc, _, _ = run([python, "-c", build_import_check_snippet(spec)], cwd=bv)
+            if rc == 0:
+                _relocate_venv_scripts(target_venv, bv)
+                log(f"baseline venv already healthy at {bv}")
+                return bv
     if not (Path(upstream) / ".git").exists():
         log(f"baseline venv: upstream {upstream} is not a git checkout")
         return None
@@ -817,8 +851,14 @@ def ensure_baseline_venv(
         env.update({key: value.replace("{commit}", baseline_commit)
                           .replace("{variant}", spec.variant)
                     for key, value in spec.install_env.items()})
+        # The worktree is deleted below. An editable install would retain a
+        # finder pointing into that deleted tree and pass only the import check
+        # run from inside it. Prefer the frozen published wheel when available.
+        requirement = (spec.install_url_template.format(
+            commit=baseline_commit, variant=spec.variant, arch=spec.arch,
+            package=spec.package) if spec.install_url_template else ".")
         rc, _, err = run([uv, "pip", "install", "--python", python,
-                          "-e", "."], cwd=wt, env=env, timeout=3600)
+                          requirement], cwd=wt, env=env, timeout=3600)
         if rc != 0:
             log(f"baseline wheel install failed: {err.strip()[-300:]}")
             shutil.rmtree(bv, ignore_errors=True)
@@ -832,6 +872,13 @@ def ensure_baseline_venv(
     finally:
         run(["git", "-C", str(upstream), "worktree", "remove",
              "--force", str(wt)], timeout=600)
+    # Probe after removal, away from both source trees: cwd must not rescue a
+    # broken install. This also protects baseline classification on later reuse.
+    rc, _, err = run([python, "-c", build_import_check_snippet(spec)], cwd=bv)
+    if rc != 0:
+        log(f"baseline venv import failed after source cleanup: {err.strip()[-200:]}")
+        shutil.rmtree(bv, ignore_errors=True)
+        return None
     log(f"baseline venv ready at {bv} (wheel {baseline_commit[:12]})")
     return bv
 

test/test_api_drift_guard.py

@@ -88,3 +88,28 @@ async def release_kv_cache_memory(self):
     assert "release_kv_cache_memory" in findings[0]
     guard.__globals__["import_class"] = lambda module, name: EngineClient if module == "upstream" else CurrentAPI
     assert guard() == []
+
+
+@pytest.mark.parametrize("keyword_only", [False, True])
+def test_guard_preserves_optional_positional_parameters(guard, keyword_only):
+    class Base:
+        def make_request_output(self, tokens, error=None):
+            pass
+
+    class KeywordOnly(Base):
+        def make_request_output(self, tokens, *, error=None):
+            pass
+
+    class Compatible(Base):
+        def make_request_output(self, tokens, error=None):
+            pass
+
+    guard.__globals__["INHERITANCE_MAP"] = [("upstream", "Base", "omni", "Omni")]
+    child = KeywordOnly if keyword_only else Compatible
+    guard.__globals__["import_class"] = lambda module, name: Base if module == "upstream" else child
+    findings = guard()
+    if keyword_only:
+        assert len(findings) == 1
+        assert "positional" in findings[0] and "error" in findings[0]
+    else:
+        assert findings == []

test/test_assembly.py

@@ -2815,3 +2815,43 @@ def test_local_timeout_multiplier_rejects_invalid_factor(factor):
     result = _local_timeout_multiplier({"local_timeout_multiplier": factor})
     assert isinstance(result, StepResult)
     assert result.failure == FailureKind.BLOCKED
+
+
+def test_local_test_job_targets_each_runtime_without_system_install(tmp_path):
+    """The same job must install into the candidate or baseline it is running."""
+    import os
+
+    from infermatrix_copilot.engine.steps.rebase_v3 import _local_test_job
+    from infermatrix_copilot.testing.runner import TestJob
+
+    binary = tmp_path / "bin"
+    binary.mkdir()
+    uv = binary / "uv"
+    uv.write_text('#!/bin/sh\nprintf "%s\\n" "$@"\n')
+    uv.chmod(0o755)
+    original = TestJob(key="attention", command="uv pip install --system --no-deps fastvideo-kernel==0.3.4",
+                       timeout_sec=1200, min_gpus=1)
+    local = _local_test_job(original, timeout_multiplier=1, virtualenv=True)
+    for runtime in ("candidate runtime", "baseline runtime"):
+        env = dict(os.environ, PATH=str(binary) + os.pathsep + os.environ["PATH"],
+                   VIRTUAL_ENV=str(tmp_path / runtime))
+        result = subprocess.run(["bash", "--norc", "-c", local.command],
+                                env=env, capture_output=True, text=True, check=True)
+        assert result.stdout.splitlines() == [
+            "pip", "install", "--python", str(tmp_path / runtime / "bin/python"),
+            "--no-deps", "fastvideo-kernel==0.3.4"]
+    assert "--system" in original.command
+    assert _local_test_job(original, timeout_multiplier=1, virtualenv=False) == original
+
+
+def test_local_timeout_budget_reaches_inner_commands(tmp_path):
+    from infermatrix_copilot.engine.steps.rebase_v3 import _local_test_job
+    from infermatrix_copilot.testing.runner import TestJob
+
+    original = TestJob(key="distributed", command="timeout 0.05s sleep 0.15\ntimeout 0.05s sleep 0.15",
+                       timeout_sec=10, min_gpus=0)
+    assert subprocess.run(["bash", "--norc", "-ec", original.command]).returncode == 124
+    local = _local_test_job(original, timeout_multiplier=10, virtualenv=True)
+    assert subprocess.run(["bash", "--norc", "-ec", local.command]).returncode == 0
+    assert local.timeout_sec == 100
+    assert original.timeout_sec == 10