diff --git a/docs/running-pipelines/cloud-debug-sessions.md b/docs/running-pipelines/cloud-debug-sessions.md index 110f8e6f..598d782f 100644 --- a/docs/running-pipelines/cloud-debug-sessions.md +++ b/docs/running-pipelines/cloud-debug-sessions.md @@ -21,6 +21,12 @@ Create a retained session by passing that script to the CLI: macrodata debug pipeline.py ``` +If the script accepts arguments, put them after `--`: + +```bash +macrodata debug pipeline.py -- --limit 10 +``` + The command executes the script locally, requires exactly one `launch_cloud(...)` call, and creates a normal cloud job from that launch. The job registers its real stage-0 shards and allocates one non-preemptible worker. @@ -44,6 +50,13 @@ macrodata debug sync pipeline.py macrodata debug run pipeline.py --max-shards 1 ``` +To replace the remembered script arguments for a sync, put the new arguments +after `--`. A later sync without `--` reuses the last remembered arguments: + +```bash +macrodata debug sync pipeline.py -- --limit 20 +``` + Sync executes the script again, captures its current cloud launch, uploads the project source and serialized stage-0 pipeline together, derives fresh shards inside the retained environment, and activates all three as one generation. @@ -89,6 +102,9 @@ macrodata debug exec pipeline.py -- python -V macrodata debug exec pipeline.py -- bash -lc 'nvidia-smi && pip freeze | head' ``` +`doctor` includes both a small `packages` summary for important runtime tools +and the complete installed distribution map in `installed_packages`. + Exec commands time out after 20 minutes by default. Use `--timeout SECONDS` for a different limit. @@ -123,10 +139,3 @@ If the retained worker crashes or Modal replaces it, the job and debug session fail rather than silently continuing in a new container. Start a new session to continue. Synchronized files, profiles, and changes made through exec are ephemeral. - -## Internal Notes - -Shard claims and completions are written to an atomic, private SQLite ledger in -the retained worker rather than the job's canonical shard ledger. Each attempt -resets that private ledger and passes it explicitly to the normal cloud worker -entrypoint. diff --git a/src/refiner/cli/debug.py b/src/refiner/cli/debug.py index e88f5136..3f9ffac5 100644 --- a/src/refiner/cli/debug.py +++ b/src/refiner/cli/debug.py @@ -76,17 +76,43 @@ def _save_profile(payload: dict[str, Any], output: str) -> Path: def _add_target(parser: argparse.ArgumentParser, *, required: bool = False) -> None: - parser.add_argument("pipeline", nargs=None if required else "?") - parser.add_argument("--job", dest="job_id") + parser.add_argument( + "pipeline", + nargs=None if required else "?", + help="Pipeline script path used to identify the debug session", + ) + parser.add_argument( + "--job", + dest="job_id", + help="Use an explicit debug job ID instead of remembered local state", + ) def _debug_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(prog="macrodata debug") subparsers = parser.add_subparsers(dest="debug_command", required=True) - create = subparsers.add_parser("create", help="Create a retained debug session") - create.add_argument("pipeline") - create.add_argument("--startup-timeout", type=int, default=1200) + create = subparsers.add_parser( + "create", + help="Create a retained debug session", + usage=( + "macrodata debug create [-h] [--startup-timeout SECONDS] " + "pipeline [-- PIPELINE_ARG ...]" + ), + epilog=( + "Pipeline arguments must follow `--`. Example:\n" + " macrodata debug create pipeline.py -- --limit 10" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + create.add_argument("pipeline", help="Pipeline script to execute and retain") + create.add_argument( + "--startup-timeout", + type=int, + default=1200, + metavar="SECONDS", + help="Maximum time to wait for the worker (default: 1200)", + ) status = subparsers.add_parser("status", help="Show debug worker status") _add_target(status) @@ -105,10 +131,28 @@ def _debug_parser() -> argparse.ArgumentParser: _add_target(profile) profile.add_argument("--output", default="refiner-debug-profile.svg") - execute = subparsers.add_parser("exec", help="Execute argv in the debug worker") + execute = subparsers.add_parser( + "exec", + help="Execute argv in the debug worker", + usage=( + "macrodata debug exec [-h] [--job JOB_ID] [--workdir PATH] " + "[--timeout SECONDS] [pipeline] -- COMMAND [ARG ...]" + ), + epilog=( + "The command and its arguments must follow `--`. Example:\n" + " macrodata debug exec pipeline.py -- python -m pip freeze" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, + ) _add_target(execute) - execute.add_argument("--workdir") - execute.add_argument("--timeout", type=int, default=1200) + execute.add_argument("--workdir", metavar="PATH", help="Worker working directory") + execute.add_argument( + "--timeout", + type=int, + default=1200, + metavar="SECONDS", + help="Maximum command runtime (default: 1200)", + ) execute.set_defaults(exec_command=[]) stop = subparsers.add_parser("stop", help="Close the debug session") @@ -116,7 +160,18 @@ def _debug_parser() -> argparse.ArgumentParser: stop.add_argument("--json", action="store_true") sync = subparsers.add_parser( - "sync", help="Synchronize source, pipeline, and private shards" + "sync", + help="Synchronize source, pipeline, and private shards", + usage=( + "macrodata debug sync [-h] [--job JOB_ID] [--json] " + "pipeline [-- PIPELINE_ARG ...]" + ), + epilog=( + "Arguments after `--` replace the remembered pipeline arguments for this sync. " + "Omit them to reuse the remembered arguments. Example:\n" + " macrodata debug sync pipeline.py -- --limit 10" + ), + formatter_class=argparse.RawDescriptionHelpFormatter, ) _add_target(sync, required=True) sync.add_argument("--json", action="store_true") @@ -256,7 +311,7 @@ def _wait_until_ready( payload = client.cloud_debug_status(job_id=job_id) status = str(payload.get("status", "unknown")) if status != last_status: - print(f"Debug worker: {status}") + print(f"Debug worker: {status}", flush=True) last_status = status if status == "ready": return payload @@ -317,7 +372,7 @@ def _cmd_create(args: argparse.Namespace) -> int: client=client, ) save_session(record) - print(f"Debug session {result.job_id} remembered for {script}") + print(f"Debug session {result.job_id} remembered for {script}", flush=True) _wait_until_ready( client=client, job_id=result.job_id, @@ -332,7 +387,8 @@ def _cmd_create(args: argparse.Namespace) -> int: ) print( f"Ready: synchronized {payload.get('files', 0)} files and " - f"{payload.get('shards', 0)} private shards" + f"{payload.get('shards', 0)} private shards", + flush=True, ) return 0 diff --git a/src/refiner/launchers/cloud.py b/src/refiner/launchers/cloud.py index f46c9c48..ec54bf44 100644 --- a/src/refiner/launchers/cloud.py +++ b/src/refiner/launchers/cloud.py @@ -607,7 +607,7 @@ def _launch(self, *, debug: bool) -> CloudLaunchResult: tracking_url=tracking_url, stage_index=resp.stage_index, ) - print(f"Cloud job launched. View job:\n {tracking_url}") + print(f"Cloud job launched. View job:\n {tracking_url}", flush=True) if debug: return CloudLaunchResult( job_id=resp.job_id, diff --git a/tests/cli/test_debug_commands.py b/tests/cli/test_debug_commands.py index 73d754ba..35bff019 100644 --- a/tests/cli/test_debug_commands.py +++ b/tests/cli/test_debug_commands.py @@ -173,6 +173,45 @@ def test_file_driven_parser_supports_create_sync_and_exec() -> None: assert execute.timeout == 10 +def _command_help(command: str, capsys) -> str: + with pytest.raises(SystemExit) as error: + debug._parse_debug_args([command, "--help"]) + assert error.value.code == 0 + return capsys.readouterr().out + + +def test_debug_help_documents_passthrough_arguments(capsys) -> None: + create_help = _command_help("create", capsys) + assert "pipeline [-- PIPELINE_ARG ...]" in create_help + assert "macrodata debug create pipeline.py -- --limit 10" in create_help + + sync_help = _command_help("sync", capsys) + assert "pipeline [-- PIPELINE_ARG ...]" in sync_help + assert "replace the remembered pipeline arguments" in sync_help + + exec_help = _command_help("exec", capsys) + assert "[pipeline] -- COMMAND [ARG ...]" in exec_help + assert "macrodata debug exec pipeline.py -- python -m pip freeze" in exec_help + + +def test_wait_until_ready_flushes_progress(monkeypatch) -> None: + client = cast(MacrodataClient, _FakeClient()) + printed: list[tuple[tuple[object, ...], dict[str, object]]] = [] + monkeypatch.setattr( + "builtins.print", + lambda *args, **kwargs: printed.append((args, kwargs)), + ) + + payload = debug._wait_until_ready( + client=client, + job_id="job-1", + timeout_secs=1200, + ) + + assert payload == {"status": "ready"} + assert printed == [(("Debug worker: ready",), {"flush": True})] + + def test_wait_until_ready_treats_canceled_as_terminal(monkeypatch) -> None: client = cast(MacrodataClient, _FakeClient()) monkeypatch.setattr(