Unified client recipes

.gitignore

@@ -166,3 +166,7 @@ batch_work_dir/
 
 # Hydra run artifacts from recipe launchers
 recipes/**/outputs/
+
+# Client recipe connection configs — these hold a Snowflake PAT. The checked-in
+# config.json.template is not matched (it does not end in .json).
+recipes/*.json

arctic_platform/client/UNIFICATION_NOTES.md

@@ -16,6 +16,9 @@ body, which job each op targets, response contract) is defined once in
 | `sft.py` | `ArcticSFTClient`, `ArcticSFTClientConfig`. |
 | `rl.py` | `ArcticRLClient`, `AsyncArcticRLClient`. |
 
+End-to-end training loops built on these frontends live outside the package, in
+`recipes/sft/standalone/` and `recipes/rl/standalone/` (Cortex only today).
+
 Import the frontends from the package root — `from arctic_platform.client import
 ArcticRLClient` — not from the module that happens to define them today.
 

arctic_platform/client/base.py

@@ -74,6 +74,19 @@ def make_transport(config: ArcticClientConfig, server_state: Any = None) -> Tran
     return HttpTransport(config)
 
 
+def _check_weight_format(config: ArcticClientConfig, weight_format: str | None) -> None:
+    """Refuse a weight_format the deployment would drop on the floor.
+
+    On-prem's ``WeightSyncRequest`` ignores unknown fields, so an unsupported
+    format would silently full-sync dense weights instead of the adapter.
+    """
+    if weight_format is not None and config.backend.type == "onprem":
+        raise ValueError(
+            f"weight_format={weight_format!r} is only supported by the remote Cortex backend; "
+            "the on-prem server always syncs full weights."
+        )
+
+
 def _maybe_print_server_profile(op: str, out: dict | None) -> None:
     """Echo the server's per-op timings when ARL_SFT_PROFILE is set; a no-op otherwise."""
     # TODO(generalize-profiling): extend profiling to every transport and workload in a
@@ -174,14 +187,19 @@ def generate(
         return self._call(generate_request(self.jobs, prompts, sampling_params, routing_key, strict))["results"]
 
     # ── weight sync + cache ──────────────────────────────────────────────
-    def sync_weights(self, cuda_ipc: bool | None = None, low_memory: bool | None = None) -> dict:
+    def sync_weights(
+        self, cuda_ipc: bool | None = None, low_memory: bool | None = None, weight_format: str | None = None
+    ) -> dict:
         """Sync training weights to sampling (staged wake → operation → wake → reset).
 
         ``cuda_ipc`` / ``low_memory`` default to the training job's ``TrainingConfig``; pass a value to override this
-        call.
+        call. ``weight_format="lora"`` broadcasts only the adapter tensors (Cortex only).
         """
+        _check_weight_format(self.config, weight_format)
         self.wake_inference(tags=["weights"])
-        out = self._call(sync_weights_request(self.jobs, cuda_ipc=cuda_ipc, low_memory=low_memory))
+        out = self._call(
+            sync_weights_request(self.jobs, cuda_ipc=cuda_ipc, low_memory=low_memory, weight_format=weight_format)
+        )
         self.wake_inference(tags=["kv_cache"])
         self.reset_prefix_cache()
         return out
@@ -266,10 +284,15 @@ async def generate(
         ]
 
     # ── weight sync + cache ──────────────────────────────────────────────
-    async def sync_weights(self, cuda_ipc: bool | None = None, low_memory: bool | None = None) -> dict:
+    async def sync_weights(
+        self, cuda_ipc: bool | None = None, low_memory: bool | None = None, weight_format: str | None = None
+    ) -> dict:
         """Async twin of ArcticClient.sync_weights (staged wake → operation → wake → reset)."""
+        _check_weight_format(self.config, weight_format)
         await self.wake_inference(tags=["weights"])
-        out = await self._acall(sync_weights_request(self.jobs, cuda_ipc=cuda_ipc, low_memory=low_memory))
+        out = await self._acall(
+            sync_weights_request(self.jobs, cuda_ipc=cuda_ipc, low_memory=low_memory, weight_format=weight_format)
+        )
         await self.wake_inference(tags=["kv_cache"])
         await self.reset_prefix_cache()
         return out

arctic_platform/client/config.py

@@ -159,6 +159,14 @@ class TrainingConfig(BaseModel):
         None,
         description="DeepSpeed worker knobs (attn_implementation, use_liger, enable_gradient_checkpointing, ...).",
     )
+    peft: dict[str, Any] | None = Field(
+        None,
+        description=(
+            "PEFT adapter config (peft_type, r, lora_alpha, lora_dropout, bias, target_modules). "
+            "None = dense fine-tuning. Applied to the training job and, when sampling is allocated, "
+            "to the sampling engine so it can serve the adapter."
+        ),
+    )
     cuda_ipc: bool = Field(
         False,
         description=(
@@ -209,6 +217,19 @@ class ArcticClientConfig(BaseModel):
     sampling_job_id: JobId | None = None
     log_prob_job_id: JobId | None = None
 
+    @model_validator(mode="after")
+    def _check_backend_supports_peft(self) -> Self:
+        # The on-prem server has no PEFT path, and to_onprem() has nowhere to put an
+        # adapter config. Silently training dense after asking for LoRA burns a run,
+        # so refuse the combination before any job or GPU is claimed.
+        if self.training.peft and self.backend.type == "onprem":
+            raise ValueError(
+                "training.peft is only supported by the remote Cortex backend; "
+                "the on-prem server trains dense only. Drop training.peft, or "
+                "switch backend to CortexConfig."
+            )
+        return self
+
     def gpus_for(self, job_type: str) -> int:
         """GPU count allocated to a job type (0 == the job type is disabled)."""
         return getattr(self, f"{job_type}_gpus")
@@ -282,6 +303,14 @@ def _cortex_training_sub_job(self) -> dict[str, Any]:
             training["model_provider"] = provider
         if worker.get("attn_implementation"):
             training["attn_implementation"] = worker["attn_implementation"]
+        # Neutrino-only engine knobs ride along in ds_worker_config; on-prem ignores them.
+        for key in ("ep_size", "mb_spec"):
+            if key in worker:
+                training[key] = worker[key]
+        if ds:
+            training["ds_config"] = ds
+        if self.training.peft:
+            training["peft_config"] = self.training.peft
         return self._cortex_sub_job("training", {"training_config": training})
 
     def _cortex_inference_sub_job(self, job_type: str, n_gpus: int) -> dict[str, Any]:
@@ -293,6 +322,8 @@ def _cortex_inference_sub_job(self, job_type: str, n_gpus: int) -> dict[str, Any
         inference: dict[str, Any] = {"max_seq_len": self.max_seq_len, "n_gpus": n_gpus}
         if vllm:
             inference["vllm_config"] = vllm
+        if self.training.peft:
+            inference["peft_config"] = self.training.peft
         return self._cortex_sub_job(job_type, {"inference_config": inference})
 
     def _cortex_sub_job(self, job_type: str, extra: dict[str, Any]) -> dict[str, Any]: