Fix interactive debugging bugs

docs/configuration.md

@@ -23,7 +23,7 @@ need to re-export `KINETIC_*` env vars each time you switch.
 | `KINETIC_OUTPUT_DIR`         | CLI + remote pod          | `gs://{bucket}/outputs/{job_id}` | Per-job durable artifact prefix. See [Checkpointing](guides/checkpointing.md).                                                                               |
 | `KINETIC_RESERVATION`        | `kinetic pool add`        | _(unset)_                        | GCP capacity reservation to consume. Pool-level config, not a per-job setting.                                                                               |
 | `KINETIC_LOG_LEVEL`          | Library                   | `INFO`                           | `DEBUG`, `INFO`, `WARNING`, `ERROR`, `FATAL`.                                                                                                                |
-| `KINETIC_DEBUG_WAIT_TIMEOUT` | Library + remote pod      | `600`                            | Seconds the remote pod waits for a debugger client to attach when `debug=True`. Applies on both sides (local `debug_attach()` and the pod's debugpy server). |
+| `KINETIC_DEBUG_WAIT_TIMEOUT` | Library + remote pod      | `600`                            | Seconds that Kinetic waits for a debugger to attach when `debug=True`. Kinetic reads the variable when you submit the job, and applies the value to the local wait and to the pod. Set the variable before you submit. The value must be a positive whole number of seconds. If the value is not valid, Kinetic uses `600`. See [Interactive Debugging](guides/debugging.md). |
 | `KINETIC_PACKAGE_ROOT`       | Library (submit)          | _(auto-detected)_                | Set this variable to the directory that Kinetic puts into `context.zip`. The directory must exist. The directory must also contain the directory that defines your function. If one of these conditions is not true, Kinetic raises a `ValueError` at submit time. See [What Ships to the Pod](guides/packaging.md). |
 | `KINETIC_NO_DEFAULT_EXCLUDES`| Library (submit)          | _(unset)_                        | Set this variable to `1` to turn the default exclusions off. Kinetic then puts `.venv`, `node_modules`, and the cache directories into `context.zip`. Kinetic always excludes `.git` and `__pycache__`.                                                             |
 | `KINETIC_CONTEXT_SIZE_WARN_MB`| Library (submit)         | `100`                            | Warning threshold in megabytes for `context.zip`. Above the threshold, Kinetic logs a warning and lists the five largest files. Set the value to `0` to turn the warning off.                                                                                    |

docs/guides/debugging.md

@@ -121,27 +121,48 @@ match.
 
 ## Path mappings and source files
 
-Kinetic fills in `pathMappings` in the printed `launch.json` so
-breakpoints set in your local files hit the matching remote files —
-no "unverified breakpoint" warnings, no file mismatch.
-
-If you attach from a directory that isn't your project root, pass
-`working_dir=` to `debug_attach()` (or replace `${workspaceFolder}` in
-the printed snippet) so the mapping points at the sources you actually
-have open.
+Your source files have the same path on the pod and on your machine.
+The runner extracts the workspace into a temporary directory. Then the
+runner makes a symbolic link at the path of your client working
+directory. That link points at the workspace.
+
+The two paths are the same, so the snippet sets `localRoot` and
+`remoteRoot` to that one directory. A breakpoint in a local file stops
+the program in the same file on the pod. VS Code does not show an
+"unverified breakpoint" warning.
+
+`kinetic jobs debug <job_id>` does not know your client working
+directory. For that command, Kinetic prints no `pathMappings` entry.
+debugpy then uses the remote paths without a change. This result is
+correct, because the paths are the same.
+
+Add a mapping only if you open your sources from a different directory
+than the directory that you submitted from. Set `localRoot` to the
+directory that you have open. Set `remoteRoot` to the directory that
+you submitted from.
 
 ## Timeouts and the attach window
 
-The pod waits up to 10 minutes for a debugger client to attach. If no
-one connects in that window, it proceeds with your function running
-normally — the job does not hang indefinitely. To extend or shorten
-that window, set `KINETIC_DEBUG_WAIT_TIMEOUT` (seconds) in your local
-environment before submitting:
+The pod waits 10 minutes for a debugger to attach. If no debugger
+attaches in that time, the pod runs your function as usual. The job
+does not wait longer than the window.
+
+To change the length of the window, set `KINETIC_DEBUG_WAIT_TIMEOUT`
+in your local environment. The unit is seconds.
 
 ```bash
 export KINETIC_DEBUG_WAIT_TIMEOUT=1800  # 30 minutes
 ```
 
+Kinetic reads the variable when you submit the job. Kinetic then puts
+the value in the pod. The client and the pod wait for the same time.
+
+Set the variable before you submit the job. A change after that time
+has no effect on a job that is already in the cluster.
+
+The value must be a positive whole number of seconds. If the value is
+not valid, Kinetic writes a warning to the log and uses 10 minutes.
+
 ## Multi-host debugging
 
 On multi-host TPU slices (Pathways backend), you attach once to the
@@ -150,6 +171,11 @@ distributed runtime doesn't start until you're ready. `jax.process_index()`
 semantics stay predictable, and you don't need to attach to each host
 separately.
 
+Kinetic gives the leader and the workers the same attach window. If
+you set `KINETIC_DEBUG_WAIT_TIMEOUT`, the new value applies to all the
+hosts. Each worker waits a short time more than the leader, so that
+normal write latency does not fail the job.
+
 :::{warning}
 **Avoid `spot=True` with `debug=True`.** Preemption mid-session
 terminates the pod, dropping your debug connection. Kinetic warns at
@@ -159,15 +185,19 @@ work.
 
 ## Automated environments
 
-`@kinetic.run(debug=True)` requires an interactive terminal — if
-`stdin` isn't a TTY (CI, `nohup`, piped input), the local client
-raises `RuntimeError` before submission so your job doesn't silently
-hang waiting for someone to attach.
+A blocking call to `@kinetic.run(debug=True)` needs an interactive
+terminal. If `stdin` is not a TTY (CI, `nohup`, or piped input), the
+client raises a `RuntimeError`. The client raises the error before it
+submits the job, so no job starts in the cluster. Without this check,
+the job waits the full window for a debugger that cannot attach. Then
+the job runs your function without a debugger.
+
+To override the check, set `KINETIC_NO_TTY_DEBUG=1`. This variable is
+for automated tests.
 
-For async submission there's no TTY requirement —
-`@kinetic.run(debug=True)` works fine in any environment, and
-`kinetic jobs debug` from an interactive shell attaches whenever
-you're ready.
+`run_async()` has no TTY requirement. Submit the job from any
+environment. Then attach with `kinetic jobs debug <job_id>` from an
+interactive shell when you are ready.
 
 ## Related pages
 

kinetic/backend/gke_client.py

@@ -13,7 +13,7 @@
 from kinetic.backend.log_streaming import LogStreamer
 from kinetic.cli.constants import KINETIC_KSA_NAME
 from kinetic.credentials import invalidate_credential_cache
-from kinetic.debug import DEBUG_WAIT_TIMEOUT, DEBUGPY_PORT
+from kinetic.debug import DEBUGPY_PORT, resolve_debug_wait_timeout
 from kinetic.job_status import JobStatus
 
 # Guards the last-seen kubeconfig context used to invalidate the
@@ -427,7 +427,7 @@ def _create_job_spec(
         client.V1EnvVar(name="PYTHONBREAKPOINT", value="debugpy.breakpoint"),
         client.V1EnvVar(
           name="KINETIC_DEBUG_WAIT_TIMEOUT",
-          value=str(DEBUG_WAIT_TIMEOUT),
+          value=str(resolve_debug_wait_timeout()),
         ),
         client.V1EnvVar(name="KINETIC_DEBUG_PORT", value=str(DEBUGPY_PORT)),
       ]

kinetic/backend/gke_client_test.py

@@ -1,5 +1,6 @@
 """Tests for kinetic.backend.gke_client — K8s job submission and monitoring."""
 
+import os
 from unittest import mock
 from unittest.mock import MagicMock
 
@@ -22,6 +23,7 @@
   GCSFUSE_CSI_DRIVER,
   GCSFUSE_VOLUMES_ANNOTATION,
 )
+from kinetic.debug import DEBUG_WAIT_TIMEOUT_ENV
 from kinetic.job_status import JobStatus
 
 
@@ -246,6 +248,40 @@ def test_fuse_single_file_mounts_parent_dir(self):
     self.assertIn("only-dir=data", vol.csi.volume_attributes["mountOptions"])
     self.assertNotIn("weights.h5", vol.csi.volume_attributes["mountOptions"])
 
+  def _debug_env(self, job):
+    container = job.spec.template.spec.containers[0]
+    return {e.name: e.value for e in container.env}
+
+  def _make_debug_job(self):
+    return _create_job_spec(
+      job_name="debug-job",
+      container_uri="img",
+      accel_config=self._make_cpu_config(),
+      job_id="j",
+      bucket_name="b",
+      namespace="ns",
+      debug=True,
+    )
+
+  def test_debug_wait_timeout_defaults_to_ten_minutes(self):
+    with mock.patch.dict(os.environ, {}, clear=False):
+      os.environ.pop(DEBUG_WAIT_TIMEOUT_ENV, None)
+      env = self._debug_env(self._make_debug_job())
+
+    self.assertEqual(env["KINETIC_DEBUG_WAIT_TIMEOUT"], "600")
+
+  def test_debug_wait_timeout_propagates_user_value_to_pod(self):
+    """A client-side override must reach the pod, or the two disagree.
+
+    The pod stops waiting at its own window. If it kept the default
+    while the user asked for longer, the pod would run the function
+    before the user finished attaching.
+    """
+    with mock.patch.dict(os.environ, {DEBUG_WAIT_TIMEOUT_ENV: "1800"}):
+      env = self._debug_env(self._make_debug_job())
+
+    self.assertEqual(env["KINETIC_DEBUG_WAIT_TIMEOUT"], "1800")
+
 
 class TestWaitForJob(absltest.TestCase):
   def setUp(self):

kinetic/backend/pathways_client.py

@@ -13,7 +13,7 @@
 from kinetic.cli.constants import KINETIC_KSA_NAME
 from kinetic.core import accelerators
 from kinetic.credentials import invalidate_credential_cache
-from kinetic.debug import DEBUG_WAIT_TIMEOUT, DEBUGPY_PORT
+from kinetic.debug import DEBUGPY_PORT, resolve_debug_wait_timeout
 from kinetic.job_status import JobStatus
 
 LWS_GROUP = "leaderworkerset.x-k8s.io"
@@ -594,6 +594,10 @@ def _create_lws_spec(
   # hang trying to join JAX's distributed runtime while the leader is
   # paused at debugpy.
   if debug:
+    # Resolved once so the leader and its workers agree on the window
+    # even if the environment changes underneath a later call.
+    wait_timeout = str(resolve_debug_wait_timeout())
+
     leader_template = copy.deepcopy(pod_template)
     leader_container = leader_template["spec"]["containers"][0]
     leader_container["env"].extend(
@@ -602,7 +606,7 @@ def _create_lws_spec(
         {"name": "PYTHONBREAKPOINT", "value": "debugpy.breakpoint"},
         {
           "name": "KINETIC_DEBUG_WAIT_TIMEOUT",
-          "value": str(DEBUG_WAIT_TIMEOUT),
+          "value": wait_timeout,
         },
         {"name": "KINETIC_DEBUG_PORT", "value": str(DEBUGPY_PORT)},
       ]
@@ -618,7 +622,7 @@ def _create_lws_spec(
         {"name": "KINETIC_DEBUG_WAIT_LEADER", "value": "1"},
         {
           "name": "KINETIC_DEBUG_WAIT_TIMEOUT",
-          "value": str(DEBUG_WAIT_TIMEOUT),
+          "value": wait_timeout,
         },
       ]
     )

kinetic/backend/pathways_client_test.py

@@ -1,5 +1,6 @@
 """Tests for kinetic.backend.pathways_client — LWS job submission and monitoring."""
 
+import os
 from unittest import mock
 from unittest.mock import MagicMock
 
@@ -26,6 +27,7 @@
 from kinetic.backend.pathways_client import (
   list_jobs as list_pathways_jobs,
 )
+from kinetic.debug import DEBUG_WAIT_TIMEOUT_ENV
 from kinetic.job_status import JobStatus
 
 _MODULE = "kinetic.backend.pathways_client"
@@ -393,6 +395,34 @@ def test_non_debug_has_no_debug_contract(self):
     self.assertNotIn("KINETIC_DEBUG", env)
     self.assertNotIn("KINETIC_DEBUG_WAIT_LEADER", env)
 
+  def _debug_wait_timeouts(self):
+    lws = self._make_spec(debug=True)["spec"]["leaderWorkerTemplate"]
+    return (
+      self._env(lws["leaderTemplate"])["KINETIC_DEBUG_WAIT_TIMEOUT"],
+      self._env(lws["workerTemplate"])["KINETIC_DEBUG_WAIT_TIMEOUT"],
+    )
+
+  def test_debug_wait_timeout_defaults_to_ten_minutes(self):
+    with mock.patch.dict(os.environ, {}, clear=False):
+      os.environ.pop(DEBUG_WAIT_TIMEOUT_ENV, None)
+      leader, worker = self._debug_wait_timeouts()
+
+    self.assertEqual(leader, "600")
+    self.assertEqual(worker, "600")
+
+  def test_debug_wait_timeout_propagates_user_value_to_both_roles(self):
+    """Leader and workers must read the same window from one resolve.
+
+    A worker waits the leader's window plus a buffer. If the two roles
+    disagreed, the workers would give up while the user was still
+    attached to the leader, and fail the job.
+    """
+    with mock.patch.dict(os.environ, {DEBUG_WAIT_TIMEOUT_ENV: "1800"}):
+      leader, worker = self._debug_wait_timeouts()
+
+    self.assertEqual(leader, "1800")
+    self.assertEqual(worker, "1800")
+
 
 class TestSubmitPathwaysJob(absltest.TestCase):
   def setUp(self):

kinetic/core/core.py

@@ -232,9 +232,10 @@ def _require_interactive_terminal():
 
   ``run(debug=True)`` blocks waiting for a VS Code debugger to attach.
   Without a TTY (CI, cron, nohup, piped input), no one can attach and
-  the job hangs for ``DEBUG_WAIT_TIMEOUT`` before falling through.
-  Fail fast with a clear message instead.  Set
-  ``KINETIC_NO_TTY_DEBUG=1`` to override (useful for automated tests).
+  the job burns the whole attach window before falling through.  Called
+  before the job is submitted so nothing lands on the cluster; fails
+  fast with a clear message instead.  Set ``KINETIC_NO_TTY_DEBUG=1`` to
+  override (useful for automated tests).
   """
   if os.environ.get("KINETIC_NO_TTY_DEBUG") == "1":
     return
@@ -291,6 +292,12 @@ def _make_decorator(
   def decorator(func):
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
+      # Checked before anything is submitted: a blocking debug call that
+      # nobody can attach to would otherwise leave a job on the cluster
+      # that waits out the whole attach window and then runs anyway.
+      if sync and debug:
+        _require_interactive_terminal()
+
       env_vars = _capture_env(capture_env_vars)
       resolved_backend = _resolve_backend_name(accelerator, backend, spot=spot)
 
@@ -334,7 +341,6 @@ def wrapper(*args, **kwargs):
 
       if sync:
         if debug:
-          _require_interactive_terminal()
           pf_proc = handle.debug_attach(working_dir=ctx.working_dir)
           try:
             return handle.result(stream_logs=False, cleanup=False)

kinetic/core/core_test.py

@@ -316,7 +316,7 @@ def test_run_debug_raises_when_stdin_not_tty(self):
       mock.patch(
         "kinetic.core.core.submit_remote",
         return_value=mock_handle,
-      ),
+      ) as mock_submit,
       mock.patch(
         "kinetic.core.core.JobContext.from_params",
         return_value=MagicMock(),
@@ -335,9 +335,43 @@ def func():
       ):
         func()
 
+      # Nothing may reach the cluster: a submitted debug job that nobody
+      # can attach to sits there for the whole attach window, then runs.
+      mock_submit.assert_not_called()
+
       # The debug attach path must not have been invoked.
       mock_handle.debug_attach.assert_not_called()
 
+  def test_run_async_debug_submits_without_tty(self):
+    """Only the blocking path needs a TTY; run_async() attaches later."""
+    mock_handle = MagicMock()
+    with (
+      mock.patch.dict(
+        os.environ,
+        _isolate_profile_env({"KINETIC_PROJECT": "proj"}),
+        clear=False,
+      ),
+      mock.patch(
+        "kinetic.core.core.submit_remote",
+        return_value=mock_handle,
+      ) as mock_submit,
+      mock.patch(
+        "kinetic.core.core.JobContext.from_params",
+        return_value=MagicMock(),
+      ),
+      mock.patch("sys.stdin.isatty", return_value=False),
+    ):
+      os.environ.pop("KINETIC_NO_TTY_DEBUG", None)
+
+      @run(accelerator="cpu", debug=True)
+      def func():
+        pass
+
+      handle = func.run_async()
+
+      self.assertIs(handle, mock_handle)
+      mock_submit.assert_called_once()
+
   def test_run_debug_allowed_when_stdin_is_tty(self):
     mock_handle = MagicMock()
     mock_handle.result.return_value = 7