feat(cli): dispatch jobs asynchronously and push results/logs to the caller

`uipath server` used to block for a whole job and leave its outcome on disk for
the caller to find. StartJob now enqueues the work and returns; the server
pushes logs while the job runs and the terminal result when it finishes.

Behaviour is a pure function of the request: a caller that supplies
`resultCallbackSocket` gets async dispatch, one that does not gets the original
blocking call, byte for byte. No handshake, no capability gate on the dispatch
path -- an older caller cannot send the field and could not serve the callback
if it did.

  POST {callback}/api/python/jobs/{jobKey}/result
  POST {callback}/api/python/jobs/{jobKey}/logs

The readiness ACK advertises protocolVersion + capabilities so the caller knows
whether logs will be forwarded before it decides to tail the file itself.

Execution stays serialised behind the process-wide lock: a job mutates process
globals (logging handlers, OTel providers, env, cwd), so queueing changes who
waits, not how many run.

Also fixes two pre-existing defects that made the API result meaningless:
_run_command_isolated hardcoded ExitCode 0 while click RETURNS ctx.exit(N)'s
code under standalone_mode=False, so every ConsoleLogger.error path reported
success; and the HTTP body now carries exitCode, which is the field the
un-upgraded .NET handler already reads.

Notes for review:
- Server diagnostics go to a stderr handle bound at import, NOT ConsoleLogger.
  ConsoleLogger resolves sys.stdout at call time, and the runtime's interceptor
  has replaced it with a writer feeding the job's execution.log -- which the
  tailer then reads and posts back. With the callback down that is a
  self-feeding loop.
- A 4xx from the callback is REJECTED, not UNREACHABLE: the caller is up and has
  moved on, so retrying cannot help and must not trip the shutdown path.
- The log tailer holds back an unterminated tail; a handler writes the record
  and only then flushes, so a poll can otherwise split one line into two
  entries that cannot be rejoined.
- Logs are tailed from the file rather than captured via a handler: the
  runtime's log interceptor strips every handler but its own, and
  uipath-runtime ships on its own release train.

Stopping a job that is already executing is deliberately NOT in this change --
it follows on top.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: robert-ursu

packages/uipath/src/uipath/_cli/_server_core.py

@@ -1,6 +1,7 @@
 """Transport-agnostic job core shared by the HTTP and uipath-ipc channels."""
 
 import asyncio
+import json
 import os
 import shlex
 from typing import Any
@@ -45,6 +46,127 @@ def parse_args(args: str | list[str] | None) -> list[str]:
     return []
 
 
+# The document is carried inline on the result push when it fits. uipath_ipc caps a
+# frame at 2 MiB and Orchestrator already spills large outputs to an attachment, so
+# anything bigger stays on disk and the caller reads the file as it always has.
+MAX_INLINE_DOCUMENT_BYTES = 1024 * 1024
+
+DEFAULT_RUNTIME_DIR = "__uipath"
+DEFAULT_RESULT_FILE = "output.json"
+DEFAULT_LOGS_FILE = "execution.log"
+
+
+def _resolve_runtime_file(
+    config_path: str, base_dir: str, key: str, default_name: str
+) -> str | None:
+    """Resolve one ``runtime.*`` file path from a uipath.json.
+
+    Mirrors UiPathRuntimeContext.from_config's ``runtime.dir`` / ``runtime.<key>``
+    mapping. ``base_dir`` anchors relative paths so this works without ever changing
+    the process's cwd.
+    """
+    if not os.path.isabs(config_path):
+        config_path = os.path.join(base_dir, config_path)
+
+    runtime: dict[str, Any] = {}
+    try:
+        with open(config_path, encoding="utf-8") as f:
+            loaded = json.load(f)
+        if isinstance(loaded, dict):
+            runtime = loaded.get("runtime") or {}
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        # Fall through to the runtime's own defaults rather than giving up. Returning
+        # None here would stop the log tailer from starting at all — and by then the
+        # caller has already been told we forward logs and stopped tailing the file
+        # itself, so the job would produce no logs anywhere.
+        runtime = {}
+
+    directory = runtime.get("dir") or DEFAULT_RUNTIME_DIR
+    name = runtime.get(key) or default_name
+    if not isinstance(directory, str) or not isinstance(name, str):
+        directory, name = DEFAULT_RUNTIME_DIR, default_name
+
+    if not os.path.isabs(directory):
+        directory = os.path.join(base_dir, directory)
+    return os.path.abspath(os.path.join(directory, name))
+
+
+def resolve_result_file_path() -> str | None:
+    """Where this job's terminal document lives. Call inside the job's env and cwd."""
+    return _resolve_runtime_file(
+        os.environ.get("UIPATH_CONFIG_PATH", "uipath.json"),
+        os.getcwd(),
+        "outputFile",
+        DEFAULT_RESULT_FILE,
+    )
+
+
+def resolve_logs_file_path(
+    env_vars: dict[str, str] | None, working_dir: str | None
+) -> str | None:
+    """Where this job will write its log file, derived from the request alone.
+
+    Pure with respect to process state: the log tailer has to know the path *before*
+    the job takes the lock and applies its env/cwd.
+    """
+    env_vars = env_vars or {}
+    base_dir = working_dir or os.getcwd()
+    config_path = env_vars.get("UIPATH_CONFIG_PATH", "uipath.json")
+    return _resolve_runtime_file(config_path, base_dir, "logsFile", DEFAULT_LOGS_FILE)
+
+
+def _read_result_document() -> tuple[str | None, str]:
+    """Return ``(document, conveyance)`` for the terminal result document.
+
+    ``conveyance`` is ``inline`` when the document rides the wire, ``file`` when the
+    caller must read it from disk (too large, unreadable, or never written).
+    """
+    path = resolve_result_file_path()
+    if not path or not os.path.exists(path):
+        return None, "file"
+
+    try:
+        if os.path.getsize(path) > MAX_INLINE_DOCUMENT_BYTES:
+            return None, "file"
+        with open(path, encoding="utf-8") as f:
+            return f.read(), "inline"
+    except (OSError, UnicodeDecodeError):
+        return None, "file"
+
+
+async def _invoke_command(cmd: Any, args: list[str]) -> dict[str, Any]:
+    """Invoke one click command and classify how it ended."""
+    try:
+        result_value = await asyncio.to_thread(cmd.main, args, standalone_mode=False)
+        # Under standalone_mode=False click RETURNS ctx.exit(N)'s code instead of
+        # raising SystemExit, so a bare int is the exit code, not a result — every
+        # ConsoleLogger.error path lands here via ctx.exit(1). The run/debug/eval
+        # callbacks only ever return a result object or None, so this is unambiguous.
+        if isinstance(result_value, int) and not isinstance(result_value, bool):
+            return {
+                "ExitCode": result_value,
+                "Error": None if result_value == 0 else f"Exit code: {result_value}",
+                "Result": None,
+                "Unexpected": False,
+            }
+        return {
+            "ExitCode": 0,
+            "Error": None,
+            "Result": result_value,
+            "Unexpected": False,
+        }
+    except SystemExit as e:
+        exit_code = e.code if isinstance(e.code, int) else 1
+        return {
+            "ExitCode": exit_code,
+            "Error": None if exit_code == 0 else f"Exit code: {exit_code}",
+            "Result": None,
+            "Unexpected": False,
+        }
+    except Exception as e:  # report any job failure as a result, not a fault
+        return {"ExitCode": 1, "Error": str(e), "Result": None, "Unexpected": True}
+
+
 async def _run_command_isolated(
     cmd: Any,
     args: list[str],
@@ -79,25 +201,13 @@ async def _run_command_isolated(
                         "ClientError": True,
                     }
 
-            result_value = await asyncio.to_thread(
-                cmd.main, args, standalone_mode=False
-            )
-            return {
-                "ExitCode": 0,
-                "Error": None,
-                "Result": result_value,
-                "Unexpected": False,
-            }
-        except SystemExit as e:
-            exit_code = e.code if isinstance(e.code, int) else 1
-            return {
-                "ExitCode": exit_code,
-                "Error": None if exit_code == 0 else f"Exit code: {exit_code}",
-                "Result": None,
-                "Unexpected": False,
-            }
-        except Exception as e:  # report any job failure as a result, not a fault
-            return {"ExitCode": 1, "Error": str(e), "Result": None, "Unexpected": True}
+            outcome = await _invoke_command(cmd, args)
+            # Must happen before the finally below restores env/cwd: the document's
+            # location comes from this job's UIPATH_CONFIG_PATH and may be relative.
+            document, conveyance = _read_result_document()
+            outcome["Document"] = document
+            outcome["DocumentConveyance"] = conveyance
+            return outcome
         finally:
             # Restore to server baseline.
             try: