fix(worker): release completed runs when children inherit output

tools/run_manage_fleet_worker.sh

@@ -924,6 +924,29 @@ completion_report_is_structured() {
     grep -q 'Demande 1' <<<"$1"
 }
 
+wait_for_provider() {
+    local provider_pid=$1
+    local stdout_logger stderr_logger exit_status=0
+    # Providers write directly to the private spool files. A spawned server
+    # may inherit those descriptors, but cannot keep a pipe to this worker
+    # open after the provider exits. Follow the exact provider PID so both
+    # log readers drain and terminate independently of descendant lifetimes.
+    tail --pid="$provider_pid" --sleep-interval=0.1 -n +1 -f -- "$output" \
+        | while IFS= read -r line || [[ -n "$line" ]]; do
+            log_provider_line "$job_id" "$line"
+        done &
+    stdout_logger=$!
+    tail --pid="$provider_pid" --sleep-interval=0.1 -n +1 -f -- "$error_output" \
+        | while IFS= read -r line || [[ -n "$line" ]]; do
+            post_job_log "$job_id" provider_stderr "$line"
+        done &
+    stderr_logger=$!
+    wait "$provider_pid" || exit_status=$?
+    wait "$stdout_logger" || true
+    wait "$stderr_logger" || true
+    return "$exit_status"
+}
+
 run_job() {
     job=$1
     job_id=$(jq -er '.id | select(type == "string" and test("^[A-Za-z0-9._-]{8,120}$"))' <<<"$job") || return
@@ -980,15 +1003,7 @@ run_job() {
                 --dangerously-bypass-approvals-and-sandbox \
                 --skip-git-repo-check \
                 -C "$cwd" \
-                - 2> >(while IFS= read -r line; do
-                    printf '%s\n' "$line" >>"$error_output"
-                    post_job_log "$job_id" provider_stderr "$line"
-                done) \
-            | tee "$output" \
-            | while IFS= read -r line; do
-                log_provider_line "$job_id" "$line"
-            done
-        provider_status=${PIPESTATUS[1]:-1}
+                - >"$output" 2>"$error_output" &
     elif [[ "$selected_provider" == jcode ]]; then
         cd -- "$cwd" || {
             set -u
@@ -1000,15 +1015,7 @@ run_job() {
                 JCODE_RUNTIME_DIR="$runtime_dir/jcode-runtime" \
                 JCODE_SERVER_EXECUTABLE="$selected_binary" \
                 "$selected_binary" --quiet --no-update --no-selfdev run --ndjson - \
-                2> >(while IFS= read -r line; do
-                    printf '%s\n' "$line" >>"$error_output"
-                    post_job_log "$job_id" provider_stderr "$line"
-                done) \
-            | tee "$output" \
-            | while IFS= read -r line; do
-                log_provider_line "$job_id" "$line"
-            done
-        provider_status=${PIPESTATUS[1]:-1}
+                >"$output" 2>"$error_output" &
     else
         cd -- "$cwd" || {
             set -u
@@ -1021,16 +1028,10 @@ run_job() {
                 --output-format stream-json \
                 --verbose \
                 --dangerously-skip-permissions \
-                2> >(while IFS= read -r line; do
-                    printf '%s\n' "$line" >>"$error_output"
-                    post_job_log "$job_id" provider_stderr "$line"
-                done) \
-            | tee "$output" \
-            | while IFS= read -r line; do
-                log_provider_line "$job_id" "$line"
-            done
-        provider_status=${PIPESTATUS[1]:-1}
+                >"$output" 2>"$error_output" &
     fi
+    wait_for_provider "$!"
+    provider_status=$?
     set -u
 
     if [[ "$selected_provider" == codex ]]; then

tools/test_manage_provider_lifecycle.py

@@ -0,0 +1,163 @@
+# SPDX-License-Identifier: Elastic-2.0
+"""Exercise the fleet worker with fake providers, never a configured service."""
+
+import json
+import os
+from pathlib import Path
+import re
+import signal
+import subprocess
+import tempfile
+import unittest
+
+
+WORKER = Path(__file__).with_name("run_manage_fleet_worker.sh").read_text()
+
+
+def function(name):
+    match = re.search(rf"^{name}\(\) \{{\n.*?^\}}$", WORKER, re.M | re.S)
+    if match is None:
+        raise AssertionError(f"missing worker function {name}")
+    return match.group(0)
+
+
+PROVIDER = r'''#!/usr/bin/env python3
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import time
+
+prompt = sys.stdin.read()
+assert "Monique completion receipt contract" in prompt
+provider = os.environ["TEST_PROVIDER"]
+if provider == "codex":
+    first = {"type": "item.started", "item": {"type": "command_execution"}}
+elif provider == "jcode":
+    first = {"type": "tool_start", "name": "fixture"}
+else:
+    first = {"type": "assistant", "message": {"content": [{"type": "text", "text": "working"}]}}
+print(json.dumps(first), flush=True)
+# A terminal-only replay of the log cannot satisfy this handshake.
+deadline = time.monotonic() + 3
+while not Path(os.environ["TEST_LOG_READY"]).exists():
+    if time.monotonic() > deadline:
+        sys.exit(42)
+    time.sleep(0.01)
+child = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
+Path(os.environ["TEST_CHILD_PID"]).write_text(str(child.pid))
+receipt = "https://github.com/example/project/issues/1#issuecomment-1"
+if provider == "codex":
+    print(json.dumps({"type": "thread.started", "thread_id": "fixture-session"}), flush=True)
+    final = {"type": "item.completed", "item": {"type": "agent_message", "text": receipt}}
+elif provider == "jcode":
+    final = {"type": "done", "session_id": "fixture-session", "text": receipt}
+else:
+    final = {"type": "result", "session_id": "fixture-session", "result": receipt}
+print(json.dumps(final), flush=True)
+sys.stderr.write("fixture diagnostic without a final newline")
+sys.stderr.flush()
+sys.exit(int(os.environ["TEST_EXIT_CODE"]))
+'''
+
+
+STUBS = r'''
+local_work_brief() { return 1; }
+workspace_for() { printf '%s\n' "$runtime_dir"; }
+report_job() { printf '%s\n' "$2" >> "$runtime_dir/reports"; }
+post_job_log() {
+    printf '%s\t%s\n' "$2" "$3" >> "$runtime_dir/logs"
+    if [[ "$2" == item.started || "$2" == tool_start || "$2" == assistant ]]; then
+        touch "$TEST_LOG_READY"
+    fi
+}
+completion_comment_permalink() { printf '%s\n' "$1"; }
+completion_comment_body() { printf 'Demande 1\nVerification: fixture\n'; }
+probe_local_auth() { return 0; }
+write_auth_health() { return 0; }
+refresh_process_snapshot() { return 0; }
+auth_failure_reason() { return 1; }
+'''
+
+
+class ProviderLifecycleTests(unittest.TestCase):
+    def run_provider(self, provider, exit_code):
+        with tempfile.TemporaryDirectory(prefix="monique-worker-test-") as directory:
+            root = Path(directory)
+            binary = root / "fake-provider"
+            binary.write_text(PROVIDER)
+            binary.chmod(0o700)
+            script = "\n".join(
+                ["set -uo pipefail", "umask 077", STUBS]
+                + [
+                    function(name)
+                    for name in (
+                        "log_provider_line",
+                        "completion_report_is_structured",
+                        "wait_for_provider",
+                        "run_job",
+                    )
+                ]
+                + ['run_job "$TEST_JOB"']
+            )
+            env = {
+                **os.environ,
+                "state_dir": directory,
+                "runtime_dir": directory,
+                "selected_provider": provider,
+                "selected_home": directory,
+                "selected_binary": str(binary),
+                "TEST_PROVIDER": provider,
+                "TEST_EXIT_CODE": str(exit_code),
+                "TEST_LOG_READY": str(root / "log-ready"),
+                "TEST_CHILD_PID": str(root / "child-pid"),
+                "TEST_JOB": json.dumps(
+                    {"id": "fixture-job-0001", "prompt": "Test the fixture", "cwd": directory}
+                ),
+            }
+            process = subprocess.Popen(
+                ["bash", "-c", script],
+                env=env,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                start_new_session=True,
+            )
+            try:
+                stdout, stderr = process.communicate(timeout=8)
+                self.assertEqual(process.returncode, 0, (stdout, stderr))
+                child = int((root / "child-pid").read_text())
+                # Completion must not depend on killing the inherited writer.
+                os.kill(child, 0)
+                expected = "done" if exit_code == 0 else "failed"
+                self.assertEqual((root / "reports").read_text().splitlines(), ["running", expected])
+                logs = (root / "logs").read_text().splitlines()
+                self.assertEqual(
+                    sum(line.startswith("provider_stderr\tfixture diagnostic") for line in logs),
+                    1,
+                )
+                output = (root / "fixture-job-0001.jsonl").read_text()
+                self.assertIn("issuecomment-1", output)
+                self.assertEqual(
+                    (root / "fixture-job-0001.stderr").read_text(),
+                    "fixture diagnostic without a final newline",
+                )
+            finally:
+                # Only the isolated fixture's process group, never a live worker.
+                try:
+                    os.killpg(process.pid, signal.SIGKILL)
+                except ProcessLookupError:
+                    pass
+                process.communicate()
+
+    def test_completed_providers_release_capacity_with_inherited_output_open(self):
+        for provider in ("codex", "jcode", "claude"):
+            with self.subTest(provider=provider):
+                self.run_provider(provider, 0)
+
+    def test_nonzero_exit_is_failed_even_when_output_contains_a_success_receipt(self):
+        self.run_provider("jcode", 7)
+
+
+if __name__ == "__main__":
+    unittest.main()