[Train 3/3 — after #78] Schedule local SFT across the Grid

README.md

@@ -660,6 +660,14 @@ grid train dataset --grid <grid-name> --out ./goal-data
 # Measure the model where it matters: independently completed held-out Goals
 grid train benchmark --suite suite.json --model <model> --run-dir ./benchmark
 
+# Queue local SFT on any compatible machine in a hosted/self-hosted grid
+grid train submit-sft --grid <grid-name> --project <id> \
+  --config grid-train.toml --data ./goal-data --backend mlx
+
+# After the job completes: fetch and verify every adapter byte before use
+grid task fetch <task-id> --grid <grid-name> --into ./trained-result
+grid train verify-result ./trained-result/grid-train-result
+
 # Turn support tickets into a reply-drafting model
 grid train init --pack support-replies
 
@@ -674,6 +682,12 @@ grid train ui
 ```
 
 - The grid samples rollouts across its nodes; one machine holds the trainer.
+- SFT jobs use the same durable task/Git plane as Goals. A matching MLX or torch node claims one;
+  if the machine disappears, its lease expires and another compatible node restarts the job from
+  the immutable input commit.
+- Training jobs have their own bounded clocks (24 hours running and seven days queued by default),
+  and successful results carry a portable SHA-256 manifest. Exit code zero without a real adapter
+  is a failed job, not a false success.
 - The LoRA adapter it produces goes back to the serving nodes under a stable name, where `auto`
   keeps routing to it.
 - Rollouts run on your own hardware: training needs the token ids and logprobs a node sampled,

cli/parser.py

@@ -1791,6 +1791,8 @@ def _add_train(sub) -> None:
         cmd_train_pull,
         cmd_train_run,
         cmd_train_schedule,
+        cmd_train_submit_sft,
+        cmd_train_verify_result,
         cmd_train_serve,
         cmd_train_sft,
         cmd_train_ui,
@@ -1953,6 +1955,37 @@ def _add_train(sub) -> None:
                      help="Where to write adapter/log/run.json (default: a timestamped folder).")
     sft.set_defaults(handler=cmd_train_sft)
 
+    submit_sft = train_sub.add_parser(
+        "submit-sft", help="Queue SFT on any compatible machine in a remote Grid"
+    )
+    submit_sft.add_argument("--grid", default=None, help="Remote grid name or id")
+    project_arg.add_project(
+        submit_sft, required=False,
+        help="Project id carrying input/results (default: your own project named 'default').")
+    submit_sft.add_argument("--config", default=None,
+                            help="Run config (default: ./grid-train.toml)")
+    submit_sft.add_argument("--data", required=True,
+                            help="sft.jsonl or a `grid train dataset` output directory")
+    submit_sft.add_argument("--backend", choices=("mlx", "torch"), required=True,
+                            help="Required worker type; explicit so scheduling is deterministic")
+    submit_sft.add_argument("--iters", type=int, default=None,
+                            help="Training iterations (MLX only)")
+    submit_sft.add_argument(
+        "--timeout-hours", type=int, default=24,
+        help="Maximum runtime after a trainer claims the job (default: 24; max: 168)")
+    submit_sft.add_argument(
+        "--queue-timeout-hours", type=int, default=168,
+        help="Maximum time to wait for a compatible trainer (default: 168; max: 720)")
+    submit_sft.add_argument("--json", action="store_true")
+    submit_sft.set_defaults(handler=cmd_train_submit_sft)
+
+    verify_result = train_sub.add_parser(
+        "verify-result", help="Verify a fetched distributed SFT adapter and its checksums"
+    )
+    verify_result.add_argument("path", help="Fetched grid-train-result directory")
+    verify_result.add_argument("--json", action="store_true")
+    verify_result.set_defaults(handler=cmd_train_verify_result)
+
     nightly = train_sub.add_parser(
         "nightly", help="One unattended cycle: train, prove it, ship it only if it won"
     )

cli/train.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import argparse
+import base64
 import dataclasses
 import json
 from pathlib import Path
@@ -549,6 +550,130 @@ def cmd_train_sft(args: argparse.Namespace) -> int:
     return 0
 
 
+def cmd_train_submit_sft(args: argparse.Namespace) -> int:
+    """Put one local SFT run on the selected Grid's durable task queue."""
+    import tomli_w
+
+    from remote import relay
+    from train.config import load_config
+    from train.sft import load_examples
+
+    from . import project_arg, remote_task
+
+    args = project_arg.resolve(args)
+
+    config_path = args.config or DEFAULT_CONFIG
+    cfg = load_config(config_path)
+    source = Path(args.data).expanduser()
+    if source.is_dir():
+        candidates = (source / "train" / "sft.jsonl", source / "sft.jsonl")
+        source = next((path for path in candidates if path.is_file()), candidates[0])
+    if not source.is_file():
+        raise SystemExit(f"grid train: no SFT dataset found at {source}")
+    # Validate every row and the existing minimum before sending bytes over the network.
+    load_examples(source)
+    data = source.read_bytes()
+
+    def portable(value):
+        if isinstance(value, tuple):
+            return [portable(item) for item in value]
+        if isinstance(value, dict):
+            return {key: portable(item) for key, item in value.items()}
+        return value
+
+    trainer = dataclasses.asdict(cfg.trainer)
+    # The explicit task run-dir is authoritative and inside the result worktree.
+    trainer["output_dir"] = ""
+    portable_config = {
+        "model": {"name": cfg.model_name},
+        "rollout": portable(dataclasses.asdict(cfg.rollout)),
+        "data": {
+            "prompts_jsonl": "grid-train-input/prompts.jsonl",
+            "verifiers_env": "",
+            "verifiers_env_args": {},
+            "learn_from_teachers": cfg.data.learn_from_teachers,
+        },
+        # ``load_config`` requires a rewards path for the shared SFT/RL schema. SFT never imports
+        # it; keep the portable file inert rather than uploading arbitrary local Python.
+        "rewards": {"python_file": "grid-train-input/rewards.py"},
+        "trainer": trainer,
+        "deploy": portable(dataclasses.asdict(cfg.deploy)),
+    }
+    config_bytes = tomli_w.dumps(portable_config).encode("utf-8")
+    reward_stub = b"# SFT does not execute reward functions.\n"
+    files = [
+        {"path": "grid-train-input/grid-train.toml",
+         "content_b64": base64.b64encode(config_bytes).decode("ascii")},
+        {"path": "grid-train-input/sft.jsonl",
+         "content_b64": base64.b64encode(data).decode("ascii")},
+        {"path": "grid-train-input/rewards.py",
+         "content_b64": base64.b64encode(reward_stub).decode("ascii")},
+    ]
+    total = len(config_bytes) + len(data) + len(reward_stub)
+    if max(len(data), len(config_bytes), len(reward_stub)) > remote_task.MAX_FILE_BYTES:
+        raise SystemExit(
+            f"grid train: each submitted file must be at most {remote_task.MAX_FILE_BYTES} bytes")
+    if total > remote_task.MAX_TOTAL_BYTES:
+        raise SystemExit(
+            f"grid train: submitted files total {total} bytes; limit is "
+            f"{remote_task.MAX_TOTAL_BYTES}")
+
+    base, token, label = remote_task._resolve(args)
+    project_id = remote_task._resolve_project(base, token, args.project)
+    timeout_hours = getattr(args, "timeout_hours", 24)
+    queue_timeout_hours = getattr(args, "queue_timeout_hours", 168)
+    if not 1 <= timeout_hours <= 168:
+        raise SystemExit("grid train: --timeout-hours must be 1-168")
+    if not 1 <= queue_timeout_hours <= 720:
+        raise SystemExit("grid train: --queue-timeout-hours must be 1-720")
+    if queue_timeout_hours <= timeout_hours:
+        raise SystemExit(
+            "grid train: --queue-timeout-hours must be greater than --timeout-hours")
+    spec = {
+        "version": 1, "backend": args.backend,
+        "config": "grid-train-input/grid-train.toml", "run_dir": "grid-train-result",
+        "run_timeout_seconds": timeout_hours * 3600,
+        "queue_timeout_seconds": queue_timeout_hours * 3600,
+    }
+    if args.iters is not None:
+        if args.backend != "mlx":
+            raise SystemExit("grid train: --iters only applies to --backend mlx")
+        if args.iters < 1 or args.iters > 10_000_000:
+            raise SystemExit("grid train: --iters must be 1-10000000")
+        spec["iters"] = args.iters
+    job = relay.create_train_job(
+        base, token, project_id=project_id, spec=spec, files=files)
+    if not isinstance(job, dict) or job.get("project_id") != project_id:
+        raise SystemExit("grid train: the relay did not confirm the requested project")
+    if args.json:
+        print(json.dumps(job, indent=2))
+        return 0
+    task_id = job.get("id")
+    print(f"SFT job queued on {label}: {task_id}")
+    print(f"  grid task follow {task_id} --grid {label}")
+    print(f"  grid task fetch {task_id} --grid {label} --into grid-train-result")
+    print("  grid train verify-result grid-train-result/grid-train-result")
+    return 0
+
+
+def cmd_train_verify_result(args: argparse.Namespace) -> int:
+    """Verify the checksums and structure of a fetched distributed SFT result."""
+    from remote.train_worker import verify_result
+
+    try:
+        result = verify_result(Path(args.path))
+    except ValueError as exc:
+        raise SystemExit(f"grid train: {exc}") from None
+    if args.json:
+        print(json.dumps(result, indent=2))
+    else:
+        print(f"Verified {args.path}")
+        print(f"  {result['backend']} adapter · {len(result['files'])} file(s) · "
+              f"{result['total_bytes']} bytes")
+        print(f"  base model: {result.get('model') or 'not recorded'}")
+    return 0
+
+
 def cmd_train_nightly(args: argparse.Namespace) -> int:
     """One unattended cycle: train, prove it on held-out work, ship it only if it won."""
     from train.config import load_config

docs/cli.md

@@ -710,6 +710,10 @@ grid train benchmark --suite <file> --model <model> --run-dir <dir> [--grid <nam
                      [--repeat <n>] [--min-pass-rate <0..1>] [--no-wait]
 
 grid train sft [--backend auto|mlx|torch] [--iters <n>] [--run-dir <dir>] [--config <path>]
+grid train submit-sft --data <sft.jsonl|dataset-dir> --backend mlx|torch
+                      [--grid <name>] [--project <id>] [--config <path>] [--iters <n>]
+                      [--timeout-hours <1..168>] [--queue-timeout-hours <1..720>] [--json]
+grid train verify-result <fetched-result-dir> [--json]
 grid train run [--config <path>]                  # the feedback loop (GRPO via TRL)
 grid train eval --run <dir> --candidate <name> [--adapter <dir>] [--base <name>] [--config <path>]
 grid train deploy --adapter <dir> [--gate] [--run <dir>] [--node <url>]... [--name <n>] [--config <path>]
@@ -781,6 +785,47 @@ production repository or call a production write tool.
 
 See [`fixtures/goal-benchmark-suite.json`](fixtures/goal-benchmark-suite.json) for the file shape.
 
+### Run SFT on the Grid
+
+`grid train submit-sft` schedules one complete SFT run on a compatible company machine. It reuses
+the existing durable task queue instead of creating a training scheduler: the config and dataset
+are committed to the selected project's Git history, a worker advertising exactly `train-mlx` or
+`train-torch` claims it, and the adapter/result directory is pushed back as the task result.
+
+```bash
+# Relay first, then enable compatible workers. Training is opt-in and is never in the default list.
+GRID_TASK_AGENT_KINDS=codex,train-mlx grid join forge --tasks-only --respawn
+
+grid train submit-sft --grid forge --project <project-id> \
+  --config grid-train.toml --data ./forge-goals --backend mlx
+grid task follow <task-id> --grid forge
+grid task fetch <task-id> --grid forge --into ./trained-result
+grid train verify-result ./trained-result/grid-train-result
+```
+
+Use `train-torch`/`--backend torch` on CUDA or other torch training machines. Install `grid[train]`
+on a training worker; it advertises no training capacity unless every dependency for its backend
+is present. The backend is explicit, so the relay never guesses from a machine name.
+
+Training jobs do not inherit the ordinary one-hour agent-task clock. The defaults are 24 hours to
+run after a trainer claims the job and seven days to wait for compatible capacity. Override them
+per job with `--timeout-hours` and `--queue-timeout-hours`; the relay and worker validate the same
+bounded values, and the queue clock must be larger. If the base model is not already cached, pass
+its registry credential explicitly with `GRID_TASK_ENV_PASSTHROUGH=HF_TOKEN`; normal worker
+environments do not leak ambient credentials into tasks.
+
+A trainer exit is not success by itself. The worker requires a non-empty adapter plus a valid
+`run.json`, rewrites worker-local paths to portable relative paths, and writes `manifest.json` with
+the size and SHA-256 of every adapter file. `grid train verify-result` rechecks that manifest after
+fetch and refuses missing, added, symlinked or changed files.
+
+This is node scheduling and failover, not cross-machine DDP: one node owns the trainer for one job.
+If that node disappears, the task lease is reclaimed and another compatible node starts from the
+same immutable input. Version 1 restarts that training attempt rather than resuming an optimizer
+checkpoint. Data and adapters cross only the selected Grid's self-hosted relay/project Git plane;
+project membership remains the authorization boundary. Deploy the Train-aware relay before
+enabling Train-aware workers.
+
 - **`grid train sft`** is imitation: it learns from the answers your team already wrote and needs
   nothing but the machine in front of you. `--backend auto` picks MLX on Apple Silicon and torch
   elsewhere.

remote/relay.py

@@ -1080,6 +1080,24 @@ def create_task(
     return task
 
 
+def create_train_job(
+    signaling_url: str,
+    access_token: str,
+    *,
+    project_id: str,
+    spec: dict[str, Any],
+    files: list[dict[str, Any]],
+) -> dict[str, Any]:
+    """Submit a typed SFT task to a Goal/Train-aware relay."""
+    return _task_oneshot(
+        signaling_url, access_token, "POST", "/relay/v1/train/jobs",
+        json={"project_id": project_id, "spec": spec, "files": files},
+        missing_route_hint=(
+            "This grid's relay does not support Grid Train jobs yet. Upgrade the relay before "
+            "enabling train-mlx or train-torch workers."),
+    )
+
+
 # Every Goal endpoint arrived as one relay feature. A CLI ahead of its relay therefore gets
 # FastAPI's bare framework 404 from any of them. "Not Found" is especially misleading for status,
 # evidence and control: it sounds as if the Goal id is wrong, when the server has never heard of a

remote/task_agent.py

@@ -845,16 +845,17 @@ def claude_available() -> bool:
 def _configured_task_harnesses() -> tuple[str, ...]:
     configured = (os.getenv("GRID_TASK_AGENT_KINDS") or "claude,codex").replace(",", " ").split()
     return tuple(dict.fromkeys(
-        kind for kind in configured if kind in ("claude", "codex")))
+        kind for kind in configured
+        if kind in ("claude", "codex", "train-mlx", "train-torch")))
 
 
 def preflight_before_serving() -> None:
     """Prove that at least one configured task harness can run before `grid join` succeeds.
 
     A task-only node is agent capacity, not specifically Claude capacity.  Codex-only company
     machines are valid Grid workers, so a missing Claude binary must not reject one that can run
-    Codex's native Goal harness.  Conversely, merely having one of the two names configured is not
-    enough: the provider may advertise only a harness whose real binary and safety checks pass.
+    Codex's native Goal harness or a local trainer. Conversely, merely configuring a name is not
+    enough: the provider advertises only a worker whose real binary/dependency checks pass.
 
     Claude uses the same `preflight`/`resolve_binary`/sandbox-package checks as its claim path.
     Codex uses `task_codex.resolve_binary`, including the measured native-Goal protocol probe.  The
@@ -867,7 +868,8 @@ def preflight_before_serving() -> None:
     configured = _configured_task_harnesses()
     if not configured:
         raise RuntimeError(
-            "GRID_TASK_AGENT_KINDS enables no supported task harness; choose codex, claude, or both")
+            "GRID_TASK_AGENT_KINDS enables no supported task harness; choose claude, codex, "
+            "train-mlx and/or train-torch")
 
     failures: list[str] = []
     causes: list[BaseException] = []
@@ -894,6 +896,21 @@ def preflight_before_serving() -> None:
             failures.append(f"codex: {str(exc) or exc.__class__.__name__}")
             causes.append(exc)
 
+    if not usable:
+        # Imported here so normal task startup never imports optional training libraries.
+        from . import train_worker
+
+        for kind in ("train-mlx", "train-torch"):
+            if kind not in configured:
+                continue
+            try:
+                train_worker.preflight(kind)
+                usable = True
+                break
+            except (Exception, SystemExit) as exc:
+                failures.append(f"{kind}: {str(exc) or exc.__class__.__name__}")
+                causes.append(exc)
+
     if usable:
         # This is shared infrastructure, not a harness opinion. Ask once after at least one harness
         # passes its own checks, and preserve the exact path-specific OSError. Checking it first