fix(cli): surface a rejected result and make IPC teardown non-raising

Two teardown-path defects on the job-api channel.

SetResult's return value was discarded, but the handler returns false when it
cannot route the job (unknown job key, or the executor threw). A rejected
result was therefore treated as delivered and silently lost; it is now
reported. The report reaches the original stderr rather than the channel that
just rejected it.

loop.close() also ran unconditionally after a timeout-bounded join. A wedged
IPC thread leaves the loop running, so close() raised: inside the context
manager's finally that masked the job's real exception (or turned a success
into exit 1), and on the connect-failure path it pre-empted the bare raise,
masking the original error. Teardown now lives in one non-raising helper used
by both paths, which only closes the loop once its thread is gone.

Author: eduard-dumitru

packages/uipath/src/uipath/_cli/_job_api.py

@@ -228,7 +228,10 @@ def _result_sink(result: Any, output_arguments_file_path: str) -> None:
             future = asyncio.run_coroutine_threadsafe(
                 callback.SetResult(job_id, dto), loop
             )
-            future.result(timeout=_SET_RESULT_TIMEOUT_S)
+            if future.result(timeout=_SET_RESULT_TIMEOUT_S) is False:
+                logger.error(
+                    "The handler rejected the job result (SetResult returned false)"
+                )
         except Exception:
             # Best-effort — a dropped result must be logged, not swallowed.
             logger.exception("Failed to deliver job result over IPC (SetResult)")
@@ -248,6 +251,24 @@ def clear_runtime_sinks() -> None:
     set_result_sink(None)
 
 
+def _stop_loop_thread(
+    loop: asyncio.AbstractEventLoop, thread: threading.Thread, timeout: float
+) -> None:
+    # Teardown must never raise: a wedged thread leaves the loop running, and close() would then
+    # raise and mask the job's own outcome (or the error that got us here).
+    try:
+        loop.call_soon_threadsafe(loop.stop)
+    except Exception:
+        pass
+    thread.join(timeout=timeout)
+    if thread.is_alive():
+        return
+    try:
+        loop.close()
+    except Exception:
+        pass
+
+
 def _new_ipc_event_loop() -> asyncio.AbstractEventLoop:
     """A fresh event loop for the connection's thread (Proactor on Windows, required for named pipes)."""
     if sys.platform == "win32":
@@ -282,9 +303,7 @@ def _shutdown(self) -> None:
             )
         except Exception:
             pass
-        self._loop.call_soon_threadsafe(self._loop.stop)
-        self._thread.join(timeout=_SET_RESULT_TIMEOUT_S)
-        self._loop.close()
+        _stop_loop_thread(self._loop, self._thread, _SET_RESULT_TIMEOUT_S)
 
 
 def connect_handler_ipc(pipe: str, job_id: str) -> _HandlerIpcConnection:
@@ -313,9 +332,7 @@ async def _build() -> Any:
         )
     except BaseException:
         # On failure, don't leak the loop/thread.
-        loop.call_soon_threadsafe(loop.stop)
-        thread.join(timeout=_SET_RESULT_TIMEOUT_S)
-        loop.close()
+        _stop_loop_thread(loop, thread, _SET_RESULT_TIMEOUT_S)
         raise
     # Install in the caller's context, not on the IPC thread: the sinks are contextvars, resolved in
     # the job's own context at teardown. The ack still runs on `loop` (a separate thread) — no deadlock.

packages/uipath/tests/cli/test_job_api.py

@@ -211,6 +211,71 @@ async def SendLog(self, jid: str, dto: Any) -> None:
     assert buf.getvalue().count("internal chatter") == 2
 
 
+def test_rejected_result_is_reported(monkeypatch):
+    captured = _fake_output_sinks(monkeypatch)
+
+    class _Callback:
+        async def SetResult(self, jid: str, dto: Any) -> bool:
+            return False
+
+    class _Result:
+        status = UiPathRuntimeStatus.SUCCESSFUL
+        error = None
+
+    records: list[logging.LogRecord] = []
+
+    class _Capture(logging.Handler):
+        def emit(self, record: logging.LogRecord) -> None:
+            records.append(record)
+
+    api_logger = logging.getLogger("uipath._cli._job_api")
+    handler = _Capture()
+    api_logger.addHandler(handler)
+    prior_level = api_logger.level
+    api_logger.setLevel(logging.ERROR)
+
+    loop = asyncio.new_event_loop()
+    thread = threading.Thread(target=loop.run_forever, daemon=True)
+    thread.start()
+    try:
+        _job_api.install_runtime_sinks("job-8", _Callback(), loop)
+        captured["sink"](_Result(), "out.args")
+    finally:
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join(timeout=5)
+        loop.close()
+        api_logger.removeHandler(handler)
+        api_logger.setLevel(prior_level)
+
+    assert any("rejected" in r.getMessage().lower() for r in records)
+
+
+def test_teardown_does_not_raise_when_the_loop_thread_is_wedged():
+    loop = asyncio.new_event_loop()
+    started = threading.Event()
+    release = threading.Event()
+
+    def _run() -> None:
+        asyncio.set_event_loop(loop)
+
+        async def _block() -> None:
+            started.set()
+            release.wait(timeout=10)
+
+        loop.run_until_complete(_block())
+
+    thread = threading.Thread(target=_run, daemon=True)
+    thread.start()
+    assert started.wait(timeout=5)
+
+    _job_api._stop_loop_thread(loop, thread, 0.2)
+    assert thread.is_alive()
+
+    release.set()
+    thread.join(timeout=5)
+    loop.close()
+
+
 def test_pending_log_sends_are_flushed_before_teardown(monkeypatch):
     _fake_output_sinks(monkeypatch)
     landed: list[str] = []