Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 16 additions & 7 deletions docs/running-pipelines/cloud-debug-sessions.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand All @@ -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:
Comment on lines +53 to +54

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Say that bare sync restores the creation arguments

When a session is created with one argument set and a sync temporarily overrides it, _cmd_sync applies that override only to the current invocation and never updates the stored DebugSessionRecord; therefore, the next bare sync reuses the original creation arguments, not the most recent override. The statement here contradicts lines 67–71 and can cause users to synchronize an unintended pipeline configuration.

AGENTS.md reference: AGENTS.md:L49-L52

Useful? React with 👍 / 👎.


```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.
Expand Down Expand Up @@ -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.

Expand Down Expand Up @@ -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.
80 changes: 68 additions & 12 deletions src/refiner/cli/debug.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -105,18 +131,47 @@ 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")
_add_target(stop)
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")
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand All @@ -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

Expand Down
2 changes: 1 addition & 1 deletion src/refiner/launchers/cloud.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
39 changes: 39 additions & 0 deletions tests/cli/test_debug_commands.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
Loading