{result.task_failure.recovery_kind === "replan_required" ? (zh ? "需要重新规划" : "Replan required") : (zh ? "需要修复原任务" : "Repair the original task")}
{result.task_failure.summary}
{zh ? "先修复任务产物或验收环境，再复核原执行。此操作复用原 Host 结果；检查通过后继续原结算，可能写回状态并扣除原任务额度。它不自动修复代码。" : "Repair the artifact or validation environment, then recheck the original Host result. Passing checks continue pending settlement, including state writeback and the original quota charge. This does not edit code."}
-
+ {result.worker_active ?
{zh ? "原执行仍在收尾。结束后选择“重新检查交付”，再复核原任务。" : "The original worker is still finishing. Choose Check delivery again after it finishes, then revalidate the task."}
: null}
+
{zh ? "原执行标识" : "Original execution identity"}{result.task_failure.turn_key}
: null}
{result.status === "accepted" && result.current_use?.state !== "unavailable" && !result.error && !result.recovery_required ?
diff --git a/docs/reference/local-delegation.md b/docs/reference/local-delegation.md
index 1c5e78a3d9..fe5f228010 100644
--- a/docs/reference/local-delegation.md
+++ b/docs/reference/local-delegation.md
@@ -293,7 +293,22 @@ Current grants, stop state and declared ancestry are checked again before
revalidation or committed-result recovery. Restoring the task output alone
does not restore unavailable source inputs.
+If the original worker still owns the operation, revalidation returns its
+current observation without starting another revalidation or settlement. A terminal
+observation can precede worker release; check delivery again after it finishes.
+Managed CLI subprocesses receive no caller stdin: their request is supplied in
+arguments, keeping MCP and other interactive input channels with their owner.
+Delegation validators also receive EOF instead of conversation input. Ordinary
+standalone caller validation retains its existing stdin inheritance.
+If a host terminates its worker process tree on disconnect (for example a
+Windows MCP Job), reconnect and use `resume` for the original operation. This
+reconciles its durable execution; it does not bypass the host's process policy.
+
中文：先修复真实产物或校验环境，再复核原执行；通过后继续原 Turn 结算，保留失败证据，不重复 Host 或扣额。停止、撤权与来源链失效仍阻止复验或提交后的恢复；仅修复输出不能恢复失效的来源输入。
+原 worker 仍占用执行时，复核只返回当前观察，不另行复核任务或结算。终态观察可能早于 worker 释放；结束后再重新检查交付。
+托管 CLI 子进程通过参数接收请求，不继承调用方的标准输入，MCP 等交互通道仍由原调用方持有。
+委派校验器同样读取 EOF，不消费对话输入；普通独立调用校验器仍保留原有的标准输入继承行为。
+宿主断连时若终止整个进程树（如 Windows MCP Job），重连后可沿原 operation 执行 `resume`，对账持久化执行，不绕过宿主进程策略。
## Source use and recheck
diff --git a/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-desktop.png b/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-desktop.png
new file mode 100644
index 0000000000..e7afd9ce17
Binary files /dev/null and b/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-desktop.png differ
diff --git a/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-mobile.png b/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-mobile.png
new file mode 100644
index 0000000000..a9a0158d30
Binary files /dev/null and b/examples/personal-workspace-browser/assets/team-task-revalidation/recovery-mobile.png differ
diff --git a/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-desktop.png b/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-desktop.png
new file mode 100644
index 0000000000..0125ad33d8
Binary files /dev/null and b/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-desktop.png differ
diff --git a/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-mobile.png b/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-mobile.png
new file mode 100644
index 0000000000..7eb12b4b17
Binary files /dev/null and b/examples/personal-workspace-browser/assets/team-task-revalidation/worker-finishing-mobile.png differ
diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json
index d1b107545d..84fad2447a 100644
--- a/loopx/canary/module_metric_baseline.json
+++ b/loopx/canary/module_metric_baseline.json
@@ -40,7 +40,7 @@
"dict_any_count": 0
},
"loopx/collaboration_mcp.py": {
- "lines": 2032
+ "lines": 2039
},
"loopx/codex_cli_probe.py": {
"any_count": 37,
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 8aa4cf24ff..e3a6ddd257 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -28,7 +28,7 @@
from mcp.server.fastmcp import FastMCP
from .file_lock import (
- exclusive_file_lock, lock_holder_host_label, lock_holder_liveness,
+ exclusive_file_lock, try_exclusive_file_lock, lock_holder_host_label, lock_holder_liveness,
LOCK_HOLDER_ABSENT, LOCK_HOLDER_DEAD, LOCK_HOLDER_RELEASED,
LOCK_HOLDER_FOREIGN_HOST, LOCK_HOLDER_LIVE, LockAcquisitionPolicy, LockAcquireTimeoutError,
)
@@ -36,6 +36,7 @@
effect_runtime_request_scope, effect_runtime_result, EffectRuntimeRemoteError,
)
from .control_plane.coordination.local_authority import local_authority_is_promoted
+from .control_plane.runtime.validation_command import noninteractive_validation
from .control_plane.turn_driver.journal_store import (
load_turn_journal,
turn_journal_path,
@@ -737,7 +738,8 @@ def start(self, binding_id: str, operation_id: str, brief: dict,
return self.read(operation_id)
def _spawn(self, operation_id: str, *, prepared_source_request_id: str | None = None) -> None:
- # Detached stdio survives requester exit; the worker owns its single-flight lock.
+ # No inherited stdio pipes. A host may still terminate its process tree;
+ # scoped resume retains the durable execution and its single-flight fence.
operation_id = require_operation_id(operation_id)
precondition = [] if prepared_source_request_id is None else ["--prepared-source-request-id", prepared_source_request_id]
subprocess.Popen([*_python_module_command("loopx.collaboration_mcp"),
@@ -800,7 +802,10 @@ def wait(self, operation_id: str) -> dict:
def revalidate(self, operation_id: str) -> dict:
"""Explicitly rerun the original postcondition; never repeat Host work."""
path = self.path(require_operation_id(operation_id))
- with exclusive_file_lock(path, policy=LockAcquisitionPolicy.SINGLE_FLIGHT):
+ with try_exclusive_file_lock(path) as acquired:
+ if acquired is None:
+ # Terminal publication can precede the original worker's release.
+ return self.read(operation_id)
if self._read_stop(path) is not None:
raise ValueError(DELEGATION_STOPPED_MESSAGE)
row = _read(path)
@@ -1318,7 +1323,8 @@ def _cli(self, binding: dict, *args: str, timeout: int = 60,
if delegated_lease is None:
completed = subprocess.run([*_python_module_command("loopx.cli"), *arguments],
cwd=binding["workspace"], capture_output=True, text=True, encoding="utf-8",
- timeout=timeout, env=environment)
+ # Requests are entirely in argv; never inherit the caller's MCP/HTTP pipe.
+ stdin=subprocess.DEVNULL, timeout=timeout, env=environment)
stdout, returncode = completed.stdout, completed.returncode
else:
from .control_plane.turn_driver.host_process_transport import run_host_process
@@ -1363,7 +1369,8 @@ def _cli(self, binding: dict, *args: str, timeout: int = 60,
return value
def _validate(self, binding: dict) -> dict:
- return delegation_validation.validate(self, binding)
+ with noninteractive_validation():
+ return delegation_validation.validate(self, binding)
def _accepted(self, binding: dict) -> dict:
return delegation_results.accepted_result(self, binding)
diff --git a/loopx/control_plane/runtime/validation_command.py b/loopx/control_plane/runtime/validation_command.py
index 675ebc7432..1e896e9b5b 100644
--- a/loopx/control_plane/runtime/validation_command.py
+++ b/loopx/control_plane/runtime/validation_command.py
@@ -3,7 +3,9 @@
import os
import shlex
import subprocess
-from collections.abc import Mapping
+from collections.abc import Iterator, Mapping
+from contextlib import contextmanager
+from contextvars import ContextVar
from pathlib import Path
from typing import Any
@@ -12,6 +14,22 @@
# Frozen wire id shared with capabilities/issue_fix. Kept verbatim for wire
# compatibility with existing caller-repo-branch receipts.
CALLER_VALIDATION_RECEIPT_SCHEMA_VERSION = "issue_fix_validation_command_v0"
+_VALIDATION_STDIN: ContextVar[int | None] = ContextVar("validation_stdin", default=None)
+
+
+@contextmanager
+def noninteractive_validation() -> Iterator[None]:
+ """Give validators EOF in this context, preserving other callers' stdin.
+
+ Trusted noninteractive hosts use this scope instead of lending their live
+ conversation input to child validators. Nested scopes and failures restore
+ the previous input policy; ordinary validation keeps inheriting stdin.
+ """
+ token = _VALIDATION_STDIN.set(subprocess.DEVNULL)
+ try:
+ yield
+ finally:
+ _VALIDATION_STDIN.reset(token)
def run_caller_validation(
@@ -49,6 +67,7 @@ def run_caller_validation(
argv,
cwd=workspace,
env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
+ stdin=_VALIDATION_STDIN.get(),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=os.name == "posix",
diff --git a/tests/control_plane/test_validation_process_cleanup.py b/tests/control_plane/test_validation_process_cleanup.py
index 553a34030a..059549904d 100644
--- a/tests/control_plane/test_validation_process_cleanup.py
+++ b/tests/control_plane/test_validation_process_cleanup.py
@@ -163,6 +163,93 @@ def test_success_keeps_inherited_stdin() -> None:
assert json.loads(proc.stdout)["passed"] is True
+def test_noninteractive_validation_restores_stdin_after_nested_exception() -> None:
+ outer = """
+import json, sys
+from pathlib import Path
+from loopx.control_plane.runtime.validation_command import noninteractive_validation, run_caller_validation
+def check(expected):
+ return run_caller_validation(Path.cwd(),
+ validation_argv=[sys.executable, '-c', f'import sys;assert sys.stdin.read()=={expected!r}'],
+ validation_label='Scoped input', timeout_seconds=3)
+with noninteractive_validation():
+ assert check('')['passed']
+ try:
+ with noninteractive_validation():
+ raise RuntimeError('synthetic nested failure')
+ except RuntimeError:
+ pass
+ assert check('')['passed']
+assert check('caller input')['passed']
+print(json.dumps({'restored': True}))
+"""
+ proc = subprocess.run(
+ [sys.executable, "-c", outer], input="caller input", text=True,
+ capture_output=True, cwd=Path(__file__).resolve().parents[2], timeout=15,
+ )
+ assert proc.returncode == 0, proc.stderr
+ assert json.loads(proc.stdout) == {"restored": True}
+
+
+def test_noninteractive_validation_does_not_change_another_thread() -> None:
+ outer = """
+import sys
+from pathlib import Path
+from concurrent.futures import ThreadPoolExecutor
+from loopx.control_plane.runtime.validation_command import noninteractive_validation, run_caller_validation
+def check(expected):
+ return run_caller_validation(Path.cwd(),
+ validation_argv=[sys.executable, '-c', f'import sys;assert sys.stdin.read()=={expected!r}'],
+ validation_label='Thread input', timeout_seconds=3)
+with ThreadPoolExecutor(max_workers=1) as pool:
+ with noninteractive_validation():
+ assert pool.submit(check, 'other thread input').result(timeout=5)['passed']
+ assert check('')['passed']
+"""
+ proc = subprocess.run(
+ [sys.executable, "-c", outer], input="other thread input", text=True,
+ capture_output=True, cwd=Path(__file__).resolve().parents[2], timeout=15,
+ )
+ assert proc.returncode == 0, proc.stderr
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows pending parent stdin regression")
+def test_noninteractive_validation_with_pending_parent_stdin_read() -> None:
+ # Keep the writer open while another parent thread waits for input. A
+ # validator must see EOF without inheriting this live conversation pipe.
+ outer = """
+import json, os, sys, threading
+from pathlib import Path
+from loopx.control_plane.runtime.validation_command import noninteractive_validation, run_caller_validation
+reading = threading.Event()
+def read():
+ reading.set()
+ sys.stdin.buffer.read(1)
+threading.Thread(target=read, daemon=True).start()
+assert reading.wait(1)
+with noninteractive_validation():
+ result = run_caller_validation(Path.cwd(),
+ validation_argv=[sys.executable, '-c', "import sys;assert sys.stdin.read()==''"],
+ validation_label='Pending parent input', timeout_seconds=3)
+print(json.dumps(result), flush=True)
+os._exit(0)
+"""
+ process = subprocess.Popen(
+ [sys.executable, "-c", outer], stdin=subprocess.PIPE,
+ stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+ cwd=Path(__file__).resolve().parents[2],
+ )
+ try:
+ # communicate() would close the writer and erase the regression.
+ assert process.wait(timeout=5) == 0
+ stdout, stderr = process.communicate(timeout=2)
+ assert json.loads(stdout)["passed"] is True, stderr
+ finally:
+ if process.poll() is None:
+ process.kill()
+ process.communicate(timeout=5)
+
+
@pytest.mark.skipif(os.name != "posix", reason="POSIX process-group regression")
@pytest.mark.parametrize("provider", ["file", "sqlite"])
def test_real_cli_timeout_keeps_canonical_todo_open_and_declaration_unchanged(
diff --git a/tests/presentation/team_task_revalidation_fixture.py b/tests/presentation/team_task_revalidation_fixture.py
index be408dc1c6..84de266a29 100644
--- a/tests/presentation/team_task_revalidation_fixture.py
+++ b/tests/presentation/team_task_revalidation_fixture.py
@@ -1,5 +1,6 @@
"""Disposable production delegation + Chat HTTP for packaged recovery validation."""
from pathlib import Path
+import hashlib
import json
import subprocess
import sys
@@ -19,7 +20,8 @@
def main():
cleanups = []
patch = pytest.MonkeyPatch()
- with tempfile.TemporaryDirectory(prefix="loopx-result-use-") as directory:
+ # Leave room for lock-incident filenames under ordinary Windows temp roots.
+ with tempfile.TemporaryDirectory(prefix="lx-rv-") as directory:
try:
root, runner = service.__wrapped__(Path(directory),
SimpleNamespace(param="file", addfinalizer=cleanups.append), patch)
@@ -32,6 +34,18 @@ def main():
runner.config = config
task_output = root / "reviewer/initial/output.json"
task_original = task_output.read_bytes()
+ terminal_written = threading.Event()
+ release_worker = threading.Event()
+ worker_errors = []
+ held_worker = None
+ observe = runner._observe
+ def observe_before_release(path, row, status, **facts):
+ observe(path, row, status, **facts)
+ if status == "rejected" and not release_worker.is_set():
+ terminal_written.set()
+ if not release_worker.wait(120):
+ raise RuntimeError("fixture worker release timed out")
+ patch.setattr(runner, "_observe", observe_before_release)
store = ChatSessionStore(runner.root)
controller = ChatRuntimeController(store=store, codex_bin="missing-codex", registry_path=runner.registry)
store.create_session(goal_id=runner.goal_id, agent_id="codex", channel_id=f"goal.{runner.goal_id}",
@@ -39,6 +53,8 @@ def main():
store.update_session("recovery", native_goal={"status": "active"}, loopx_mode={"settings": {
"agent_id": "lead", "execution_config": ".loopx/config/delegations.json"}})
server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler)
+ # Drain this fixture's request handlers before retiring their runtime.
+ server.daemon_threads = False
server.verbose = False
server.registry_path = runner.registry
server.runtime_root = runner.root
@@ -50,10 +66,29 @@ def main():
try:
for command in sys.stdin:
action = command.strip()
- if action == "task-failure":
+ if action == "task-failure-active":
task_output.write_text("{}")
- runner.start("failure", "failure-1", brief())
- assert wait(runner, "failure-1")["status"] == "rejected"
+ def execute():
+ try:
+ runner.execute("failure-1")
+ except BaseException as error:
+ worker_errors.append(error)
+ def spawn(operation_id):
+ nonlocal held_worker
+ assert operation_id == "failure-1"
+ held_worker = threading.Thread(target=execute, daemon=True)
+ held_worker.start()
+ with pytest.MonkeyPatch.context() as worker_patch:
+ worker_patch.setattr(runner, "_spawn", spawn)
+ runner.start("failure", "failure-1", brief())
+ assert terminal_written.wait(120), worker_errors
+ current = runner.read("failure-1")
+ assert current["status"] == "rejected" and current["worker_active"]
+ elif action == "task-release":
+ release_worker.set()
+ held_worker.join(timeout=30)
+ assert not held_worker.is_alive() and not worker_errors, worker_errors
+ assert not wait(runner, "failure-1")["worker_active"]
elif action == "task-repair":
task_output.write_bytes(task_original)
elif action == "task-commit-response-lost":
@@ -71,12 +106,16 @@ def lose_response(binding, *args, **kwargs):
assert runner.read("failure-1")["recovery_required"] is True
elif action == "task-inspect":
print(json.dumps({"operation": runner.read("failure-1"),
+ "record_sha256": hashlib.sha256(runner.path("failure-1").read_bytes()).hexdigest(),
"host_calls": (root / "reviewer/initial/host-invocations").read_text()}), flush=True)
continue
elif action != "inspect":
raise ValueError("unknown fixture command")
print(json.dumps({"ok": True}), flush=True)
finally:
+ release_worker.set()
+ if held_worker is not None:
+ held_worker.join(timeout=30)
server.shutdown()
server.server_close()
thread.join(timeout=5)
diff --git a/tests/test_delegation_result_use.py b/tests/test_delegation_result_use.py
index 6935c896aa..8b97dc42e0 100644
--- a/tests/test_delegation_result_use.py
+++ b/tests/test_delegation_result_use.py
@@ -163,7 +163,8 @@ def accepted_chain(root, runner):
def linked_request(source, workspace):
artifact = source["artifacts"][0]
path = workspace / "declared-input.json"
- path.write_text(artifact["text"])
+ # A receiver must retain the accepted bytes, including line endings.
+ path.write_bytes(artifact["text"].encode("utf-8"))
return {**brief(), "inputs": [{"ref": path.name, "description": "Exact accepted source",
"sha256": artifact["sha256"], "delegation": {"operation_id": source["operation_id"],
"ref": artifact["ref"], "relation": "uses"}}]}
diff --git a/tests/test_delegation_task_revalidation.py b/tests/test_delegation_task_revalidation.py
index be60c355e4..a532c7e838 100644
--- a/tests/test_delegation_task_revalidation.py
+++ b/tests/test_delegation_task_revalidation.py
@@ -4,6 +4,8 @@
import os
import sys
import subprocess
+from concurrent.futures import ThreadPoolExecutor
+from threading import Event
import pytest
from mcp import ClientSession, StdioServerParameters
@@ -15,6 +17,93 @@
service = delegation_service
+def test_active_rejected_worker_defers_revalidation_until_release(service, monkeypatch):
+ """Terminal publication is observable before the original worker releases ownership."""
+ from loopx.control_plane.turn_driver.journal_store import turn_journal_path
+
+ root, runner = service
+ output = root / "analyst/initial/output.json"
+ saved = output.read_bytes()
+ output.write_text("{}")
+ published, release = Event(), Event()
+ observe = runner._observe
+
+ def pause_after_rejection(path, row, status, **facts):
+ observe(path, row, status, **facts)
+ if status == "rejected":
+ published.set()
+ assert release.wait(120), "test did not release the terminal worker"
+
+ with ThreadPoolExecutor(max_workers=1) as pool:
+ with monkeypatch.context() as worker_patch:
+ worker_patch.setattr(runner, "_spawn", lambda _: None)
+ worker_patch.setattr(runner, "_observe", pause_after_rejection)
+ runner.start("analysis", "analysis-1", brief())
+ worker = pool.submit(runner.execute, "analysis-1")
+ try:
+ assert published.wait(120), "worker did not publish the task failure"
+ failed = runner.read("analysis-1")
+ assert failed["status"] == "rejected" and failed["worker_active"]
+ turn_key = failed["task_failure"]["turn_key"]
+ path = runner.path("analysis-1")
+ journal = turn_journal_path(runner.root, goal_id=runner.goal_id, turn_key=turn_key)
+ original_record, original_journal = path.read_bytes(), journal.read_bytes()
+ with monkeypatch.context() as duplicate_patch:
+ def forbidden_effect(*args, **kwargs):
+ pytest.fail("an active operation must not retry validation or write execution state")
+ duplicate_patch.setattr(runner, "_cli", forbidden_effect)
+ duplicate_patch.setattr(runner, "_spawn", forbidden_effect)
+ duplicate_patch.setattr(runner, "_fenced_write", forbidden_effect)
+ duplicate = runner.revalidate("analysis-1")
+ assert duplicate["operation_id"] == "analysis-1"
+ assert duplicate["task_failure"] == failed["task_failure"]
+ assert duplicate["worker_active"] is True
+ assert path.read_bytes() == original_record
+ assert journal.read_bytes() == original_journal
+ assert (root / "analyst/initial/host-invocations").read_text().strip() == "1"
+
+ read = runner.read
+ def release_after_active_read(operation):
+ result = read(operation)
+ if result["status"] == "rejected" and result["worker_active"]:
+ release.set()
+ return result
+ with monkeypatch.context() as readiness_patch:
+ readiness_patch.setattr(runner, "read", release_after_active_read)
+ ready = wait(runner)
+ assert ready["worker_active"] is False
+ finally:
+ release.set()
+ worker.result(timeout=30)
+
+ output.write_bytes(saved)
+ runner.revalidate("analysis-1")
+ accepted = wait(runner)
+ assert accepted["status"] == "accepted", accepted
+ assert json.loads(runner.path("analysis-1").read_text())["turn_key"] == turn_key
+ assert (root / "analyst/initial/host-invocations").read_text().strip() == "1"
+
+
+def test_revalidation_preserves_an_internal_lock_timeout(service, monkeypatch):
+ """Only operation-lock contention is a no-op; retry execution failures stay visible."""
+ from loopx.file_lock import LockAcquireTimeoutError
+
+ root, runner = service
+ output = root / "analyst/initial/output.json"
+ saved = output.read_bytes()
+ output.write_text("{}")
+ runner.start("analysis", "analysis-1", brief())
+ assert wait(runner)["status"] == "rejected"
+ output.write_bytes(saved)
+ failure = LockAcquireTimeoutError(incident={}, incident_recorded=False, incident_channel="fixture")
+ def fail_inside_retry(*args, **kwargs):
+ raise failure
+ monkeypatch.setattr(runner, "_cli", fail_inside_retry)
+ with pytest.raises(LockAcquireTimeoutError) as raised:
+ runner.revalidate("analysis-1")
+ assert raised.value is failure
+
+
def test_original_task_revalidation_repairs_output_without_repeating_host(service):
root, runner = service
output = root / "analyst/initial/output.json"
@@ -45,6 +134,9 @@ async def recheck_mcp():
result = await session.call_tool("revalidate_delegation", {"operation_id": "analysis-1"})
assert not result.isError, result
asyncio.run(recheck_mcp())
+ # A Windows MCP host may close its entire Job on disconnect. Reconcile
+ # the original committed Turn through the existing independent caller.
+ runner.resume("analysis-1")
accepted = wait(runner)
assert accepted["status"] == "accepted", json.dumps({"read": accepted, "record": json.loads(runner.path("analysis-1").read_text())})
assert "task_failure" not in accepted
diff --git a/tests/test_local_delegation.py b/tests/test_local_delegation.py
index e2bba4c897..44c64121f5 100644
--- a/tests/test_local_delegation.py
+++ b/tests/test_local_delegation.py
@@ -73,7 +73,7 @@ def retire_runtime():
"from pathlib import Path; import sys; "
"from loopx.control_plane.effect_runtime import _runtime_dir, restart_effect_runtime; "
"assert _runtime_dir().parent == Path(sys.argv[1]); "
- "result = restart_effect_runtime(); "
+ "result = restart_effect_runtime(wait_for_exit=True); "
"assert result['status'] in {'stopped', 'not_running'}, result",
str(tmp_path),
], cwd=Path(__file__).resolve().parents[1], env=runtime_env,
@@ -264,8 +264,9 @@ def wait(service, operation="analysis-1"):
while time.monotonic() < deadline:
result = service.read(operation)
if result["status"] in {"accepted", "rejected"}:
- return result
- if result.get("error"):
+ if not result["worker_active"]:
+ return result
+ elif result.get("error"):
pytest.fail(str(result))
time.sleep(0.25)
pytest.fail(str(service.read(operation)))