diff --git a/mkdocs.yml b/mkdocs.yml
index 62d7feabf1..a2d6f3aa31 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -297,6 +297,7 @@ nav:
- Services: docs/concepts/services.md
- Volumes: docs/concepts/volumes.md
- More:
+ - Endpoints: docs/concepts/endpoints.md
- Gateways: docs/concepts/gateways.md
- Secrets: docs/concepts/secrets.md
- Projects: docs/concepts/projects.md
@@ -344,6 +345,7 @@ nav:
- dev-environment: docs/reference/dstack.yml/dev-environment.md
- task: docs/reference/dstack.yml/task.md
- service: docs/reference/dstack.yml/service.md
+ - endpoint: docs/reference/dstack.yml/endpoint.md
- fleet: docs/reference/dstack.yml/fleet.md
- gateway: docs/reference/dstack.yml/gateway.md
- volume: docs/reference/dstack.yml/volume.md
@@ -366,6 +368,7 @@ nav:
- dstack volume: docs/reference/cli/dstack/volume.md
- dstack gateway: docs/reference/cli/dstack/gateway.md
- dstack secret: docs/reference/cli/dstack/secret.md
+ - dstack endpoint: docs/reference/cli/dstack/endpoint.md
- dstack export: docs/reference/cli/dstack/export.md
- dstack import: docs/reference/cli/dstack/import.md
- HTTP API:
diff --git a/mkdocs/docs/concepts/endpoints.md b/mkdocs/docs/concepts/endpoints.md
new file mode 100644
index 0000000000..f058e8bfc0
--- /dev/null
+++ b/mkdocs/docs/concepts/endpoints.md
@@ -0,0 +1,154 @@
+---
+title: Endpoints
+description: Creating and reusing optimized model inference recipes
+---
+
+# Endpoints
+
+An endpoint configuration is a new `dstack` feature that lets you use an agent to create presets: validated and optimized model inference endpoint recipes. Once a preset is created, it can be reused to deploy model inference on validated hardware without an agent.
+
+The value of presets comes from combining two fundamental features: agent-driven model inference optimization and the `dstack` [service](services.md) primitive, which can deploy model inference to any cloud, Kubernetes, or on-prem cluster.
+
+> The endpoints feature is experimental and may change.
+
+??? info "Prerequisites"
+ Before using endpoint presets, make sure you’ve [installed](../installation.md) the server and CLI, and created a [fleet](fleets.md).
+
+ Creating an endpoint preset requires the `claude` CLI to be installed on the machine where you create a preset.
+
+## Define an endpoint
+
+Before you can create or reuse an endpoint preset, you first have to define an endpoint configuration. The filename must end with `.dstack.yml`.
+
+
+
+```yaml
+type: endpoint
+name: qwen25-7b
+
+model:
+ base: Qwen/Qwen2.5-7B-Instruct
+
+env:
+ - HF_TOKEN
+```
+
+
+
+Since `base` is specified, the preset can use any compatible variant of the base model, including a different precision, quantization, or trusted fork.
+
+If you want to deploy an exact model, set `model` directly to the repo of that model:
+
+
+
+```yaml
+type: endpoint
+name: qwen25-7b
+
+model: Qwen/Qwen2.5-7B-Instruct
+
+env:
+ - HF_TOKEN
+```
+
+
+
+Set `context_length` to require a minimum context length. Placement properties,
+including `fleets`, `backends`, `max_price`, and `spot_policy`, constrain both
+creation and reuse. Environment variables such as `HF_TOKEN` can be passed
+through `env`.
+
+See the [reference](../reference/dstack.yml/endpoint.md) for all supported configuration options.
+
+## Create a preset
+
+To create a preset, pass the configuration file to the `dstack endpoint preset create` command:
+
+
+
+```shell
+$ dstack endpoint preset create -f endpoint.dstack.yml
+```
+
+
+
+This command executes entirely locally and uses the locally installed `claude` CLI along with `dstack`'s bundled skills. The agent uses a `dstack` task to find the best serving recipe matching the available fleet offers. Once the recipe is found, it submits a `dstack` service for a final benchmark. The validated recipe is saved locally under `~/.dstack/presets`.
+
+??? info "Claude authorization"
+ Preset creation supports two Claude authorization methods. To use an Anthropic API key, set:
+
+ ```shell
+ export DSTACK_AGENT_ANTHROPIC_API_KEY=...
+ ```
+
+ If you are already logged in with `claude`, use the existing authorization instead:
+
+ ```shell
+ export DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH=1
+ ```
+
+ These options are mutually exclusive.
+
+## List presets
+
+Use `dstack endpoint preset` to list existing presets:
+
+
+
+```shell
+$ dstack endpoint preset list
+ MODEL GPU
+ Qwen/Qwen2.5-7B-Instruct
+ recipe=8f3a12c4 L4:24GB:1..
+```
+
+
+
+Each base model corresponds to one preset. A preset may contain multiple recipes built for specific hardware. Each recipe includes benchmark data.
+
+Pass `-v` to include the full service resources and verified context length.
+
+## Apply a preset
+
+To deploy a preset as a service, pass the endpoint configuration to the `dstack endpoint preset apply` command:
+
+
+
+```shell
+$ dstack endpoint preset apply -f endpoint.dstack.yml
+```
+
+
+
+If you don't pass `--recipe ID` (or specify it in the endpoint configuration), `dstack` automatically picks one of the recipes that matches the available fleet offers.
+
+If a recipe matches, `dstack` deploys it as a service.
+
+## Delete a preset
+
+You can delete a specific recipe or the entire preset.
+
+
+
+```shell
+dstack endpoint preset delete --recipe 8f3a12c4
+```
+
+
+
+To delete the entire preset and all its recipes, pass the base model:
+
+
+
+```shell
+dstack endpoint preset delete Qwen/Qwen2.5-7B-Instruct
+```
+
+
+
+For command options and agent settings, see the
+[`dstack endpoint` CLI reference](../reference/cli/dstack/endpoint.md).
+
+!!! info "What's next?"
+ 1. Learn how dstack [services](services.md) work
+ 2. Learn how to configure [fleets](fleets.md)
diff --git a/mkdocs/docs/reference/cli/dstack/endpoint.md b/mkdocs/docs/reference/cli/dstack/endpoint.md
new file mode 100644
index 0000000000..ddfc328b20
--- /dev/null
+++ b/mkdocs/docs/reference/cli/dstack/endpoint.md
@@ -0,0 +1,80 @@
+# dstack endpoint
+
+The `dstack endpoint` commands create, list, apply, and delete local endpoint
+[preset recipes](../../../concepts/endpoints.md).
+
+## dstack endpoint preset list
+
+The `dstack endpoint preset list` command lists locally stored recipes.
+
+##### Usage
+
+
+
+```shell
+$ dstack endpoint preset list --help
+#GENERATE#
+```
+
+
+
+## dstack endpoint preset create
+
+The `dstack endpoint preset create` command uses an agent to create and save a
+verified recipe from an endpoint configuration.
+
+##### Usage
+
+
+
+```shell
+$ dstack endpoint preset create --help
+#GENERATE#
+```
+
+
+
+##### Agent settings
+
+Set either `DSTACK_AGENT_ANTHROPIC_API_KEY` or
+`DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH` before creating a preset.
+
+| Variable | Description |
+| --- | --- |
+| `DSTACK_AGENT_ANTHROPIC_API_KEY` | Anthropic API key used by the agent. |
+| `DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH` | Use the existing `claude` login. |
+| `DSTACK_AGENT_CLAUDE_PATH` | `claude` executable name or path. Defaults to `claude` from `PATH`. |
+| `DSTACK_AGENT_ANTHROPIC_MODEL` | Claude model used by the agent. |
+| `DSTACK_AGENT_CLAUDE_EFFORT` | Claude effort level: `low`, `medium`, `high`, `xhigh`, or `max`. |
+
+## dstack endpoint preset apply
+
+The `dstack endpoint preset apply` command selects a matching local recipe and
+submits its service.
+
+##### Usage
+
+
+
+```shell
+$ dstack endpoint preset apply --help
+#GENERATE#
+```
+
+
+
+## dstack endpoint preset delete
+
+The `dstack endpoint preset delete` command deletes one local recipe by ID or
+all recipes for a base model.
+
+##### Usage
+
+
+
+```shell
+$ dstack endpoint preset delete --help
+#GENERATE#
+```
+
+
diff --git a/mkdocs/docs/reference/dstack.yml/endpoint.md b/mkdocs/docs/reference/dstack.yml/endpoint.md
new file mode 100644
index 0000000000..77b83d8954
--- /dev/null
+++ b/mkdocs/docs/reference/dstack.yml/endpoint.md
@@ -0,0 +1,53 @@
+# `endpoint`
+
+The `endpoint` configuration type describes a model request and the constraints
+used to create or apply an [endpoint preset](../../concepts/endpoints.md).
+
+## Root reference
+
+#SCHEMA# dstack._internal.core.models.endpoints.EndpointConfiguration
+ overrides:
+ show_root_heading: false
+ type:
+ required: true
+
+### `model`
+
+=== "Base model"
+
+ Allows the creation agent to select a compatible model variant.
+
+ #SCHEMA# dstack._internal.core.models.endpoints.EndpointModelBase
+ overrides:
+ show_root_heading: false
+
+=== "Exact model"
+
+ Requires an exact model repo or path and optionally sets another
+ client-facing model name.
+
+ #SCHEMA# dstack._internal.core.models.endpoints.EndpointModelRepo
+ overrides:
+ show_root_heading: false
+
+### `retry`
+
+#SCHEMA# dstack._internal.core.models.profiles.ProfileRetry
+ overrides:
+ show_root_heading: false
+
+### `utilization_policy`
+
+#SCHEMA# dstack._internal.core.models.profiles.UtilizationPolicy
+ overrides:
+ show_root_heading: false
+ type:
+ required: true
+
+### `schedule`
+
+#SCHEMA# dstack._internal.core.models.profiles.Schedule
+ overrides:
+ show_root_heading: false
+ type:
+ required: true
diff --git a/pyproject.toml b/pyproject.toml
index b61f974ed3..f0373de1b1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -62,11 +62,19 @@ artifacts = [
"src/dstack/_internal/server/statics/**",
]
+[tool.hatch.build.targets.sdist.force-include]
+"skills/dstack/SKILL.md" = "skills/dstack/SKILL.md"
+"skills/dstack-prototyping/SKILL.md" = "skills/dstack-prototyping/SKILL.md"
+
[tool.hatch.build.targets.wheel]
artifacts = [
"src/dstack/_internal/server/statics/**",
]
+[tool.hatch.build.targets.wheel.force-include]
+"skills/dstack/SKILL.md" = "dstack/_internal/core/resources/endpoint_agent/skills/dstack/SKILL.md"
+"skills/dstack-prototyping/SKILL.md" = "dstack/_internal/core/resources/endpoint_agent/skills/dstack-prototyping/SKILL.md"
+
[tool.hatch.metadata.hooks.fancy-pypi-readme]
content-type = "text/markdown"
diff --git a/skills/dstack-prototyping/SKILL.md b/skills/dstack-prototyping/SKILL.md
new file mode 100644
index 0000000000..4ad8033d3f
--- /dev/null
+++ b/skills/dstack-prototyping/SKILL.md
@@ -0,0 +1,100 @@
+---
+name: dstack-prototyping
+description: |
+ Use with the dstack skill for model-serving work when the image, serving command, resources, backend/fleet choice, or service behavior is not proven. Guides task-first prototyping on real hardware, choosing fleets/backends that can reuse idle instances and caches, checking vLLM/SGLang sources, and verifying the final dstack service with a model request.
+---
+
+# dstack Prototyping
+
+Use `/dstack` for CLI commands, YAML fields, apply/attach behavior, service URLs,
+and other dstack syntax. This skill explains how to use dstack runs while the
+model-serving recipe is still unknown.
+
+## Goal
+
+Find a working dstack service recipe for the requested model.
+
+Before submitting a service, use a task on real hardware to test the serving
+image, install/runtime assumptions, model download, cache path, command, port,
+launch flags, resources, env vars, backend/fleet choice, and local model
+request. Then submit the same recipe as a service and verify the model through
+the dstack service URL.
+
+## Choose Where To Run
+
+Choose only VM-based backends, SSH fleets, or Kubernetes fleets because they support idle instances and/or instance volumes. That lets later runs reuse the provisioned/idle instance or instance volumes used by runs for caching model weights (and possibly other writes). You must follow this rule even if there are fleets/backends/offers that are cheaper. The only exception from this rule is when the required GPU class (regardless of the price) is not available through VM-based backend, SSH fleet, or Kubernetes fleet.
+
+Read `https://dstack.ai/docs/concepts/backends.md` to know exactly which
+backends are VM-based.
+
+## Check Serving Sources
+
+Check serving-framework sources early enough to choose the image, command,
+launch flags, resources, cache paths, request format, and expected model
+behavior.
+
+For vLLM and SGLang, use these as credible sources:
+
+- vLLM recipes and model index: `https://recipes.vllm.ai/` and
+ `https://recipes.vllm.ai/models.json`
+- vLLM recipe docs: `https://docs.vllm.ai/projects/recipes/en/stable/`
+- SGLang docs and cookbook: `https://docs.sglang.ai/` and
+ `https://lmsysorg.mintlify.app/cookbook/intro`
+
+Use deeper serving-engine writeups, such as
+`https://www.lmsys.org/blog/2026-07-02-agent-assisted-sglang-development`, when
+recipes and docs do not explain the model, hardware, or serving failure.
+
+## Use A Task Before Service
+
+Before submitting a service, start a long-lived task:
+
+```yaml
+commands:
+ - sleep infinity
+```
+
+or an equivalent idle command.
+
+Submit the task detached, attach or SSH into it when available, and run commands
+inside the live environment. Test the image, installs, model download and cache
+path, serving command, port, launch flags, local model request, and expected
+model behavior.
+
+When starting a long-running command in the background from a non-interactive
+SSH command, use `nohup`, redirect stdin from `/dev/null`, and redirect
+stdout/stderr to a log file so the SSH command returns while the process keeps
+running. For example (the command can be any long-running command):
+
+```shell
+nohup vllm serve ... /tmp/vllm.log 2>&1 &
+```
+
+If the image, hardware choice, or major install path changes, submit another
+task so the changed setup is tested before service verification.
+
+Do not move to a service after checking only GPU visibility, imports, logs, or a
+health endpoint. Start the server inside the task and send a request that uses
+the requested model. For a chat or reasoning model, check the response behavior
+the endpoint is expected to support, such as reasoning output when that model is
+supposed to expose it.
+
+Follow `/dstack` structured status guidance when polling task or service status.
+After requesting a task or service stop before another submission, wait until
+that run reaches a terminal status. This allows dstack to reuse its instance or
+instance volumes when available.
+
+## Verify As A Service
+
+Submit the service after the task has verified the recipe: image, command, port,
+resources, env vars, cache mounts if used, backend/fleet choice, and model
+request.
+
+Use the service as a duplicate check of the same recipe under dstack service
+runtime. The model request that worked locally in the task must also work
+through the dstack service URL.
+
+If service verification fails because the image, install, model download,
+command, resources, cache, or model behavior needs to change, go back to a task.
+If the recipe is still right and only the service config is wrong, fix the
+service config and submit the service again.
diff --git a/skills/dstack/SKILL.md b/skills/dstack/SKILL.md
index 4df1843b75..556317cb67 100644
--- a/skills/dstack/SKILL.md
+++ b/skills/dstack/SKILL.md
@@ -34,7 +34,7 @@ submits and exits.
1) Show plan: `echo "n" | dstack apply -f `
2) If plan is OK and user confirms, apply detached: `dstack apply -f -y -d`
-3) Check status once: `dstack ps -v`
+3) Check the run: `dstack run get --json`
4) If dev-environment or task with ports and running: attach to surface IDE link/ports/SSH alias (agent runs attach in background); ask to open link
5) If attach fails in sandbox: request escalation; if not approved, ask the user to run `dstack attach` locally and share the output
@@ -76,7 +76,15 @@ submits and exits.
- `dstack apply` without `-d` for runs
- `dstack ps -w`
-Agents should avoid blocking: use `-d`, timeouts, or background attach. When attach is needed, run it in the background by default (`nohup ...`), but describe it to the user simply as "attach" unless they ask for a live foreground session. Prefer `dstack ps -v` and poll in a loop if the user wants to watch status.
+Agents should avoid blocking: use `-d`, timeouts, or background attach. When attach is needed, run it in the background by default (`nohup ...`), but describe it to the user simply as "attach" unless they ask for a live foreground session.
+
+When waiting programmatically for a specific run, use
+`dstack run get --json` and read its top-level `status`. Run statuses
+are `pending`, `submitted`, `provisioning`, `running`, `terminating`,
+`terminated`, `failed`, and `done`; the last three are terminal. Stop waiting
+when the run reaches the state needed for the next action or a terminal status.
+Never parse or grep human-readable `dstack ps` output; its status column may
+display a job message such as `no offers`.
**All other commands:** Use 10-60s timeout. Most complete within this range. **While waiting, monitor the output** - it may contain errors, warnings, or prompts requiring attention.
@@ -94,9 +102,9 @@ Agents should avoid blocking: use `-d`, timeouts, or background attach. When att
After submitting a run with `-d` (dev-environment, task, service), first determine whether submission failed. If the apply output shows errors (validation, no offers, etc.), stop and surface the error.
-If the run was submitted, do a quick status check with `dstack ps -v`, then guide the user through relevant next steps:
+If the run was submitted, check it with `dstack run get --json`, then guide the user through relevant next steps:
If you need to prompt for next actions, be explicit about the dstack step and command (avoid vague questions). When speaking to the user, refer to the action as "attach" (not "background attach").
-- **Monitor status:** Report the current status (provisioning/pulling/running/finished) and offer to keep watching. Poll `dstack ps -v` every 10-20s if the user wants updates.
+- **Monitor status:** Report the current status and offer to keep watching. If watching, poll `dstack run get --json` every 10-20 seconds until it reaches the state needed for the next action or a terminal status.
- **Attach when running:** For agents, run attach in the background by default so the session does not block. Use it to capture IDE links/SSH alias or enable port forwarding; when describing the action to the user, just say "attach".
- **Dev environments or tasks with ports:** Once `running`, attach to surface the IDE link/port forwarding/SSH alias, then ask whether to open the IDE link. Never open links without explicit approval.
- **Services:** Prefer using service endpoints. Attach only if the user explicitly needs port forwarding or full log replay.
@@ -368,7 +376,7 @@ domain: example.com
4. **Verify apply status:**
```bash
- dstack ps -v
+ dstack run get --json
```
**Workflow for infrastructure (fleet, volume, gateway):**
@@ -533,7 +541,7 @@ Common issues:
- **No offers:** Check `dstack offer`; if submitting a run, ensure at least one fleet can provision or reuse matching instances
- **No fleet:** Ensure at least one fleet is created
- **Configuration errors:** Validate YAML syntax; check `dstack apply` output for specific errors
-- **Provisioning timeouts:** Use `dstack ps -v` to see provisioning status; consider spot vs on-demand
+- **Provisioning timeouts:** Inspect the run with `dstack run get --json`; consider spot vs on-demand
- **Connection issues:** Verify server status, check authentication, ensure network access to backends
**When errors occur:**
diff --git a/src/dstack/_internal/cli/commands/endpoint.py b/src/dstack/_internal/cli/commands/endpoint.py
new file mode 100644
index 0000000000..3137cfc4ed
--- /dev/null
+++ b/src/dstack/_internal/cli/commands/endpoint.py
@@ -0,0 +1,219 @@
+import argparse
+import os
+
+from argcomplete import FilesCompleter # type: ignore[attr-defined]
+
+from dstack._internal.cli.commands import BaseCommand
+from dstack._internal.cli.services.completion import ProjectNameCompleter
+from dstack._internal.cli.services.endpoint_preset_apply import apply_endpoint_preset
+from dstack._internal.cli.services.endpoint_preset_create import create_endpoint_preset
+from dstack._internal.cli.services.endpoint_presets import (
+ EndpointPresetStore,
+ load_endpoint_configuration,
+)
+from dstack._internal.cli.utils.common import confirm_ask, console
+from dstack._internal.cli.utils.endpoint_presets import print_endpoint_presets
+from dstack._internal.core.errors import CLIError, ConfigurationError
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.envs import EnvSentinel
+from dstack._internal.core.services import is_valid_dstack_resource_name
+from dstack.api import Client
+
+
+class EndpointCommand(BaseCommand):
+ NAME = "endpoint"
+ DESCRIPTION = "Manage model inference endpoints"
+
+ def _register(self) -> None:
+ self._parser.add_argument(
+ "--project",
+ help="The name of the project. Defaults to [code]$DSTACK_PROJECT[/]",
+ metavar="NAME",
+ default=os.getenv("DSTACK_PROJECT"),
+ ).completer = ProjectNameCompleter() # type: ignore[attr-defined]
+ self._parser.set_defaults(subfunc=lambda _: self._parser.print_help())
+ subparsers = self._parser.add_subparsers(dest="action")
+ preset_parser = subparsers.add_parser(
+ "preset",
+ help="Manage endpoint presets",
+ formatter_class=self._parser.formatter_class,
+ )
+ preset_parser.set_defaults(subfunc=self._list)
+ preset_subparsers = preset_parser.add_subparsers(dest="preset_action")
+
+ list_parser = preset_subparsers.add_parser(
+ "list",
+ help="List endpoint presets",
+ formatter_class=self._parser.formatter_class,
+ )
+ list_parser.add_argument("-v", "--verbose", action="store_true")
+ list_parser.set_defaults(subfunc=self._list)
+
+ create_parser = preset_subparsers.add_parser(
+ "create",
+ help="Create an endpoint preset",
+ formatter_class=self._parser.formatter_class,
+ )
+ _add_configuration_args(create_parser)
+ create_parser.add_argument(
+ "--keep-service",
+ action="store_true",
+ help="Leave the verified service running",
+ )
+ create_parser.set_defaults(subfunc=self._create)
+
+ apply_parser = preset_subparsers.add_parser(
+ "apply",
+ help="Apply an endpoint preset",
+ formatter_class=self._parser.formatter_class,
+ )
+ _add_configuration_args(apply_parser)
+ apply_parser.add_argument("--recipe", metavar="ID", help="The recipe ID to use")
+ apply_parser.add_argument(
+ "-y", "--yes", action="store_true", help="Do not ask for confirmation"
+ )
+ apply_parser.add_argument(
+ "--force", action="store_true", help="Force apply when no changes are detected"
+ )
+ apply_parser.add_argument(
+ "-d", "--detach", action="store_true", help="Exit after submitting the service"
+ )
+ apply_parser.add_argument(
+ "-v", "--verbose", action="store_true", help="Show all plan properties"
+ )
+ apply_parser.set_defaults(subfunc=self._apply)
+
+ delete_parser = preset_subparsers.add_parser(
+ "delete",
+ help="Delete an endpoint preset or recipe",
+ formatter_class=self._parser.formatter_class,
+ )
+ delete_target = delete_parser.add_mutually_exclusive_group(required=True)
+ delete_target.add_argument(
+ "base",
+ nargs="?",
+ metavar="BASE",
+ help="The base model whose preset to delete",
+ )
+ delete_target.add_argument(
+ "--recipe",
+ metavar="ID",
+ help="Delete one recipe by ID",
+ )
+ delete_parser.add_argument(
+ "-y", "--yes", action="store_true", help="Do not ask for confirmation"
+ )
+ delete_parser.set_defaults(subfunc=self._delete)
+
+ def _command(self, args: argparse.Namespace) -> None:
+ super()._command(args)
+ try:
+ args.subfunc(args)
+ except KeyboardInterrupt:
+ console.print("\nOperation interrupted by user. Exiting...")
+ exit(0)
+
+ def _list(self, args: argparse.Namespace) -> None:
+ print_endpoint_presets(
+ EndpointPresetStore().list(),
+ verbose=getattr(args, "verbose", False),
+ )
+
+ def _create(self, args: argparse.Namespace) -> None:
+ _, configuration = load_endpoint_configuration(args.configuration_file)
+ _apply_name(configuration, args.name)
+ _resolve_endpoint_env(configuration)
+ result = create_endpoint_preset(
+ api=Client.from_config(project_name=args.project),
+ configuration=configuration,
+ store=EndpointPresetStore(),
+ keep_service=args.keep_service,
+ )
+ console.print(
+ f"Endpoint preset recipe [code]{result.recipe.id}[/] for "
+ f"[code]{result.recipe.base}[/] saved to [code]{result.path}[/]"
+ )
+ if args.keep_service:
+ console.print(f"Final service [code]{result.final_run_name}[/] kept running")
+
+ def _apply(self, args: argparse.Namespace) -> None:
+ configuration_path, configuration = load_endpoint_configuration(args.configuration_file)
+ _apply_name(configuration, args.name)
+ apply_endpoint_preset(
+ api=Client.from_config(project_name=args.project),
+ configuration=configuration,
+ configuration_path=configuration_path,
+ recipe_id=args.recipe or configuration.recipe,
+ command_args=args,
+ store=EndpointPresetStore(),
+ )
+
+ def _delete(self, args: argparse.Namespace) -> None:
+ store = EndpointPresetStore()
+ recipe = None
+ if args.recipe is not None:
+ recipe = store.get(args.recipe)
+ if recipe is None:
+ raise CLIError(f"Endpoint preset recipe {args.recipe!r} does not exist")
+ message = (
+ f"Delete endpoint preset recipe [code]{recipe.id}[/] for [code]{recipe.base}[/]?"
+ )
+ else:
+ recipes = [recipe for recipe in store.list() if recipe.base == args.base]
+ if not recipes:
+ raise CLIError(f"Endpoint preset {args.base!r} does not exist")
+ message = (
+ f"Delete endpoint preset [code]{args.base}[/] and its "
+ f"{len(recipes)} recipe{'s' if len(recipes) != 1 else ''}?"
+ )
+ if not args.yes and not confirm_ask(message):
+ console.print("\nExiting...")
+ return
+ if args.recipe is not None:
+ assert recipe is not None
+ store.delete_recipe(args.recipe)
+ console.print(
+ f"Endpoint preset recipe [code]{recipe.id}[/] for [code]{recipe.base}[/] deleted"
+ )
+ else:
+ count = store.delete_preset(args.base)
+ console.print(
+ f"Endpoint preset [code]{args.base}[/] deleted "
+ f"({count} recipe{'s' if count != 1 else ''})"
+ )
+
+
+def _add_configuration_args(parser: argparse.ArgumentParser) -> None:
+ parser.add_argument(
+ "-f",
+ "--file",
+ required=True,
+ metavar="FILE",
+ dest="configuration_file",
+ help="The endpoint configuration file",
+ ).completer = FilesCompleter(allowednames=["*.yml", "*.yaml"]) # type: ignore[attr-defined]
+ parser.add_argument(
+ "-n",
+ "--name",
+ metavar="NAME",
+ help="The endpoint name. Required when the configuration omits name",
+ )
+
+
+def _apply_name(configuration: EndpointConfiguration, name: str | None) -> None:
+ if name is not None:
+ configuration.name = name
+ if configuration.name is None:
+ raise CLIError("Endpoint name is required. Set `name` in the configuration or use --name")
+ if not is_valid_dstack_resource_name(configuration.name):
+ raise CLIError("Endpoint name must match '^[a-z][a-z0-9-]{1,40}$'")
+
+
+def _resolve_endpoint_env(configuration: EndpointConfiguration) -> None:
+ for key, value in configuration.env.items():
+ if not isinstance(value, EnvSentinel):
+ continue
+ try:
+ configuration.env[key] = value.from_env(os.environ)
+ except ValueError as e:
+ raise ConfigurationError(str(e)) from e
diff --git a/src/dstack/_internal/cli/main.py b/src/dstack/_internal/cli/main.py
index 32f15a95f8..335f7693f1 100644
--- a/src/dstack/_internal/cli/main.py
+++ b/src/dstack/_internal/cli/main.py
@@ -8,6 +8,7 @@
from dstack._internal.cli.commands.attach import AttachCommand
from dstack._internal.cli.commands.completion import CompletionCommand
from dstack._internal.cli.commands.delete import DeleteCommand
+from dstack._internal.cli.commands.endpoint import EndpointCommand
from dstack._internal.cli.commands.event import EventCommand
from dstack._internal.cli.commands.export import ExportCommand
from dstack._internal.cli.commands.fleet import FleetCommand
@@ -67,6 +68,7 @@ def main():
ApplyCommand.register(subparsers)
AttachCommand.register(subparsers)
DeleteCommand.register(subparsers)
+ EndpointCommand.register(subparsers)
EventCommand.register(subparsers)
ExportCommand.register(subparsers)
FleetCommand.register(subparsers)
diff --git a/src/dstack/_internal/cli/services/configurators/run.py b/src/dstack/_internal/cli/services/configurators/run.py
index 16b0f0a87b..8c6ac510a5 100644
--- a/src/dstack/_internal/cli/services/configurators/run.py
+++ b/src/dstack/_internal/cli/services/configurators/run.py
@@ -6,8 +6,9 @@
import subprocess
import sys
import time
+from dataclasses import dataclass
from pathlib import Path
-from typing import Dict, List, Optional, Set, TypeVar
+from typing import Dict, Generic, List, Optional, Set, TypeVar
import gpuhunt
from pydantic import parse_obj_as
@@ -54,9 +55,10 @@
TaskConfiguration,
)
from dstack._internal.core.models.repos import RepoHeadWithCreds
+from dstack._internal.core.models.repos.base import Repo
from dstack._internal.core.models.repos.remote import RemoteRepo, RemoteRepoCreds
from dstack._internal.core.models.resources import CPUSpec
-from dstack._internal.core.models.runs import JobStatus, JobSubmission, RunSpec, RunStatus
+from dstack._internal.core.models.runs import JobStatus, JobSubmission, RunPlan, RunSpec, RunStatus
from dstack._internal.core.services.diff import diff_models
from dstack._internal.core.services.repos import get_repo_creds_and_default_branch
from dstack._internal.core.services.ssh.ports import PortUsedError
@@ -80,6 +82,13 @@
RunConfigurationT = TypeVar("RunConfigurationT", bound=AnyRunConfiguration)
+@dataclass(frozen=True)
+class PreparedRunConfiguration(Generic[RunConfigurationT]):
+ configuration: RunConfigurationT
+ repo: Repo
+ run_plan: RunPlan
+
+
class BaseRunConfigurator(
ApplyEnvVarsConfiguratorMixin,
BaseApplyConfigurator[RunConfigurationT],
@@ -91,6 +100,23 @@ def apply_configuration(
command_args: argparse.Namespace,
configurator_args: argparse.Namespace,
):
+ prepared = self.prepare_configuration(
+ conf=conf,
+ configuration_path=configuration_path,
+ configurator_args=configurator_args,
+ )
+ return self.apply_prepared_configuration(
+ prepared=prepared,
+ command_args=command_args,
+ configurator_args=configurator_args,
+ )
+
+ def prepare_configuration(
+ self,
+ conf: RunConfigurationT,
+ configuration_path: str,
+ configurator_args: argparse.Namespace,
+ ) -> PreparedRunConfiguration[RunConfigurationT]:
if configurator_args.repo and configurator_args.no_repo:
raise CLIError("Either --repo or --no-repo can be specified")
@@ -121,6 +147,21 @@ def apply_configuration(
profile=profile,
ssh_identity_file=configurator_args.ssh_identity_file,
)
+ return PreparedRunConfiguration(
+ configuration=conf,
+ repo=repo,
+ run_plan=run_plan,
+ )
+
+ def apply_prepared_configuration(
+ self,
+ prepared: PreparedRunConfiguration[RunConfigurationT],
+ command_args: argparse.Namespace,
+ configurator_args: argparse.Namespace,
+ ):
+ conf = prepared.configuration
+ repo = prepared.repo
+ run_plan = prepared.run_plan
no_fleets = False
if len(run_plan.job_plans[0].offers) == 0:
@@ -721,7 +762,7 @@ def register_args(cls, parser: argparse.ArgumentParser):
super().register_args(parser)
cls.register_commands_args(parser)
- def apply_args(self, conf: TaskConfiguration, args: argparse.Namespace):
+ def apply_args(self, conf: ServiceConfiguration, args: argparse.Namespace):
super().apply_args(conf, args)
self.apply_commands_args(conf, args)
diff --git a/src/dstack/_internal/cli/services/endpoint_agent_runtime.py b/src/dstack/_internal/cli/services/endpoint_agent_runtime.py
new file mode 100644
index 0000000000..b0f5a51006
--- /dev/null
+++ b/src/dstack/_internal/cli/services/endpoint_agent_runtime.py
@@ -0,0 +1,602 @@
+import asyncio
+import json
+import os
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+from contextlib import contextmanager, suppress
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Iterator, Optional, Sequence
+
+import psutil
+from rich.text import Text
+
+from dstack._internal.cli.utils.common import console
+from dstack._internal.compat import IS_WINDOWS
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.endpoint_agent import AGENT_FINAL_REPORT_JSON_SCHEMA
+from dstack.api import Client
+
+_SKILL_NAMES = ("dstack", "dstack-prototyping")
+_PROGRESS_FILENAME = "progress.jsonl"
+_SUBMISSIONS_FILENAME = "submissions.jsonl"
+_BENCHMARKS_FILENAME = "benchmarks.jsonl"
+_FINAL_REPORT_FILENAME = "final_report.json"
+_PROGRESS_ENV = "DSTACK_ENDPOINT_PROGRESS_LOG"
+_REDACTION = "[redacted]"
+_CLAUDE_TOOLS = "Bash,Read,Write,Edit,WebFetch,WebSearch,StructuredOutput"
+_CLAUDE_EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
+_MAX_RUN_NAME_LENGTH = 41
+_MAX_UNIX_SOCKET_PATH_BYTES = 103
+_INHERITED_ENV_NAMES = (
+ "PATH",
+ "HOME",
+ "USER",
+ "SSL_CERT_FILE",
+ "REQUESTS_CA_BUNDLE",
+ "HTTP_PROXY",
+ "HTTPS_PROXY",
+ "NO_PROXY",
+ "http_proxy",
+ "https_proxy",
+ "no_proxy",
+)
+_WINDOWS_INHERITED_ENV_NAMES = (
+ "APPDATA",
+ "COMSPEC",
+ "HOMEDRIVE",
+ "HOMEPATH",
+ "LOCALAPPDATA",
+ "PATHEXT",
+ "PROGRAMDATA",
+ "SYSTEMDRIVE",
+ "SYSTEMROOT",
+ "USERPROFILE",
+ "USERNAME",
+ "WINDIR",
+)
+_SENSITIVE_INHERITED_ENV_NAMES = (
+ "HTTP_PROXY",
+ "HTTPS_PROXY",
+ "NO_PROXY",
+ "http_proxy",
+ "https_proxy",
+ "no_proxy",
+)
+
+
+@dataclass(frozen=True)
+class EndpointAgentWorkspace:
+ path: Path
+ dstack_home: Path
+
+ @property
+ def temp_path(self) -> Path:
+ return self.path / ".tmp"
+
+ @property
+ def bin_path(self) -> Path:
+ return self.path / "bin"
+
+ @property
+ def progress_path(self) -> Path:
+ return self.path / _PROGRESS_FILENAME
+
+ @property
+ def submissions_path(self) -> Path:
+ return self.path / _SUBMISSIONS_FILENAME
+
+ @property
+ def benchmarks_path(self) -> Path:
+ return self.path / _BENCHMARKS_FILENAME
+
+ @property
+ def final_report_path(self) -> Path:
+ return self.path / _FINAL_REPORT_FILENAME
+
+ @property
+ def stdout_path(self) -> Path:
+ return self.path / "agent_stdout.jsonl"
+
+ @property
+ def stderr_path(self) -> Path:
+ return self.path / "agent_stderr.jsonl"
+
+
+@dataclass(frozen=True)
+class ClaudeAuth:
+ api_key: Optional[str]
+ executable: str
+ effort: Optional[str]
+ model: str
+ use_existing: bool
+
+
+@dataclass
+class EndpointAgentProcessOutput:
+ report_data: Optional[dict[str, Any]] = None
+ error: Optional[str] = None
+
+
+@contextmanager
+def endpoint_agent_workspace() -> Iterator[EndpointAgentWorkspace]:
+ with tempfile.TemporaryDirectory(prefix="dpe-", dir=_get_short_temp_dir()) as directory:
+ root = Path(directory)
+ _validate_control_socket_path(root)
+ workspace = EndpointAgentWorkspace(path=root / "w", dstack_home=root / "h")
+ _prepare_workspace(workspace)
+ yield workspace
+
+
+def get_claude_auth() -> ClaudeAuth:
+ api_key = os.getenv("DSTACK_AGENT_ANTHROPIC_API_KEY") or None
+ use_existing = _get_bool_env("DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH")
+ if api_key and use_existing:
+ raise CLIError(
+ "DSTACK_AGENT_ANTHROPIC_API_KEY and "
+ "DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH cannot both be set"
+ )
+ if not api_key and not use_existing:
+ raise CLIError(
+ "DSTACK_AGENT_ANTHROPIC_API_KEY is not set. Set it or opt in to existing "
+ "Claude CLI auth with DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH=1"
+ )
+ configured_path = os.getenv("DSTACK_AGENT_CLAUDE_PATH") or "claude"
+ executable = shutil.which(configured_path)
+ if executable is None:
+ raise CLIError(f"Claude executable not found: {configured_path}")
+ effort = os.getenv("DSTACK_AGENT_CLAUDE_EFFORT") or None
+ if effort is not None and effort not in _CLAUDE_EFFORT_LEVELS:
+ raise CLIError(
+ f"DSTACK_AGENT_CLAUDE_EFFORT must be one of: {', '.join(_CLAUDE_EFFORT_LEVELS)}"
+ )
+ return ClaudeAuth(
+ api_key=api_key,
+ executable=executable,
+ effort=effort,
+ model=os.getenv("DSTACK_AGENT_ANTHROPIC_MODEL", "claude-opus-4-8"),
+ use_existing=use_existing,
+ )
+
+
+def build_endpoint_agent_env(
+ *,
+ api: Client,
+ endpoint_env: dict[str, str],
+ auth: ClaudeAuth,
+ workspace: EndpointAgentWorkspace,
+ token: str,
+) -> dict[str, str]:
+ env = {name: value for name in _INHERITED_ENV_NAMES if (value := os.getenv(name))}
+ env.update(endpoint_env)
+ if IS_WINDOWS:
+ env.update(
+ {name: value for name in _WINDOWS_INHERITED_ENV_NAMES if (value := os.getenv(name))}
+ )
+ env["PATH"] = os.pathsep.join([str(workspace.bin_path), env.get("PATH", "")])
+ env["DSTACK_SERVER_URL"] = api.client.base_url
+ env["DSTACK_PROJECT"] = api.project
+ env["DSTACK_TOKEN"] = token
+ env["DSTACK_ENDPOINT_SERVER_URL"] = api.client.base_url
+ env["DSTACK_ENDPOINT_BEARER_TOKEN"] = token
+ env[_PROGRESS_ENV] = str(workspace.progress_path)
+ for name in ["TMPDIR", "TEMP", "TMP"]:
+ env[name] = str(workspace.temp_path)
+ if auth.api_key is not None:
+ env["ANTHROPIC_API_KEY"] = auth.api_key
+ env["HOME"] = str(workspace.dstack_home)
+ if IS_WINDOWS:
+ env["USERPROFILE"] = str(workspace.dstack_home)
+ else:
+ env["HOME"] = str(Path.home())
+ if IS_WINDOWS:
+ env["USERPROFILE"] = str(Path.home())
+ return env
+
+
+async def run_endpoint_agent(
+ *,
+ prompt: str,
+ env: dict[str, str],
+ workspace: EndpointAgentWorkspace,
+ auth: ClaudeAuth,
+ redacted_values: Sequence[str],
+) -> EndpointAgentProcessOutput:
+ command = _prepare_subprocess_command(_build_claude_command(auth=auth))
+ progress_tailer = _ProgressTailer(
+ path=workspace.progress_path,
+ redacted_values=redacted_values,
+ )
+ progress_task = asyncio.create_task(progress_tailer.run())
+ proc: Optional[asyncio.subprocess.Process] = None
+ try:
+ proc = await asyncio.create_subprocess_exec(
+ *command,
+ cwd=workspace.path,
+ env=env,
+ stdin=asyncio.subprocess.PIPE,
+ stdout=asyncio.subprocess.PIPE,
+ stderr=asyncio.subprocess.PIPE,
+ start_new_session=not IS_WINDOWS,
+ )
+ assert proc.stdin is not None
+ assert proc.stdout is not None
+ assert proc.stderr is not None
+ proc.stdin.write(prompt.encode())
+ with suppress(BrokenPipeError, ConnectionResetError):
+ await proc.stdin.drain()
+ proc.stdin.close()
+ stdout_task = asyncio.create_task(
+ _read_process_stream(
+ stream=proc.stdout,
+ path=workspace.stdout_path,
+ parse_result=True,
+ redacted_values=redacted_values,
+ )
+ )
+ stderr_task = asyncio.create_task(
+ _read_process_stream(
+ stream=proc.stderr,
+ path=workspace.stderr_path,
+ parse_result=False,
+ redacted_values=redacted_values,
+ )
+ )
+ stdout_output, stderr_output, returncode = await asyncio.gather(
+ stdout_task,
+ stderr_task,
+ proc.wait(),
+ )
+ except BaseException:
+ if proc is not None and proc.returncode is None:
+ await _terminate_process(proc)
+ raise
+ finally:
+ progress_task.cancel()
+ with suppress(asyncio.CancelledError):
+ await progress_task
+ progress_tailer.flush()
+
+ output = stdout_output
+ if output.report_data is None:
+ output.report_data = stderr_output.report_data
+ if output.error is None:
+ output.error = stderr_output.error
+ if output.report_data is None and returncode != 0:
+ output.error = output.error or f"Claude exited with return code {returncode}"
+ return output
+
+
+def print_endpoint_progress(message: str) -> None:
+ timestamp = datetime.now(timezone.utc).astimezone().strftime("%Y-%m-%d %H:%M:%S")
+ console.print(
+ Text(f"[{timestamp}]", style="log.time"),
+ Text(message.rstrip("\r\n"), style="log.message"),
+ soft_wrap=True,
+ )
+
+
+def get_redacted_values(values: Sequence[str]) -> tuple[str, ...]:
+ return tuple(sorted({value for value in values if value}, key=len, reverse=True))
+
+
+def contains_redacted_value(value: Any, redacted_values: Sequence[str]) -> bool:
+ if isinstance(value, str):
+ return any(
+ value == redacted or (len(redacted) >= 8 and redacted in value)
+ for redacted in redacted_values
+ )
+ if isinstance(value, dict):
+ return any(contains_redacted_value(item, redacted_values) for item in value.values())
+ if isinstance(value, list):
+ return any(contains_redacted_value(item, redacted_values) for item in value)
+ return False
+
+
+def get_sensitive_inherited_env_values() -> list[str]:
+ return [value for name in _SENSITIVE_INHERITED_ENV_NAMES if (value := os.getenv(name))]
+
+
+def redact(value: str, redacted_values: Sequence[str]) -> str:
+ for redacted_value in redacted_values:
+ value = value.replace(redacted_value, _REDACTION)
+ return value
+
+
+def _get_bool_env(name: str) -> bool:
+ value = os.getenv(name)
+ if value is None:
+ return False
+ normalized = value.strip().lower()
+ if normalized in {"1", "true", "yes", "on"}:
+ return True
+ if normalized in {"0", "false", "no", "off", ""}:
+ return False
+ raise CLIError(f"{name} must be a boolean")
+
+
+def _validate_control_socket_path(build_root: Path) -> None:
+ if IS_WINDOWS:
+ return
+ path = build_root / "h" / ".dstack" / "ssh" / f"{'x' * _MAX_RUN_NAME_LENGTH}.control.sock"
+ if len(os.fsencode(path)) > _MAX_UNIX_SOCKET_PATH_BYTES:
+ raise CLIError(f"Temporary path is too long for an SSH control socket: {build_root}")
+
+
+def _prepare_workspace(workspace: EndpointAgentWorkspace) -> None:
+ workspace.path.mkdir(mode=0o700, parents=True, exist_ok=False)
+ workspace.dstack_home.mkdir(mode=0o700)
+ workspace.temp_path.mkdir(mode=0o700)
+ for path in [
+ workspace.progress_path,
+ workspace.submissions_path,
+ workspace.benchmarks_path,
+ ]:
+ path.touch()
+ workspace.bin_path.mkdir()
+ _install_python_command(workspace.bin_path, "progress", _get_progress_script())
+ (workspace.dstack_home / ".ssh").mkdir(mode=0o700)
+ _install_dstack_wrapper(workspace.bin_path, workspace.dstack_home)
+ _install_home_wrapper(workspace.bin_path, "ssh", workspace.dstack_home)
+ _install_skills(workspace.path)
+
+
+def _install_dstack_wrapper(bin_dir: Path, home: Path) -> None:
+ script = f"""#!{sys.executable}
+import os
+import sys
+
+from dstack._internal.cli.main import main
+
+os.environ["HOME"] = {json.dumps(str(home))}
+if os.name == "nt":
+ os.environ["USERPROFILE"] = {json.dumps(str(home))}
+sys.argv[0] = "dstack"
+raise SystemExit(main())
+"""
+ _install_python_command(bin_dir, "dstack", script)
+
+
+def _install_home_wrapper(bin_dir: Path, command: str, home: Path) -> None:
+ executable = shutil.which(command)
+ if executable is None:
+ script = f"""#!{sys.executable}
+import sys
+
+print("Endpoint preset creation could not find the {command} executable.", file=sys.stderr)
+raise SystemExit(127)
+"""
+ else:
+ script = f"""#!{sys.executable}
+import os
+import subprocess
+import sys
+
+os.environ["HOME"] = {json.dumps(str(home))}
+if os.name == "nt":
+ os.environ["USERPROFILE"] = {json.dumps(str(home))}
+raise SystemExit(subprocess.call([{json.dumps(executable)}, *sys.argv[1:]]))
+"""
+ _install_python_command(bin_dir, command, script)
+
+
+def _install_python_command(bin_dir: Path, name: str, script: str) -> None:
+ if not IS_WINDOWS:
+ path = bin_dir / name
+ path.write_text(script, encoding="utf-8")
+ path.chmod(0o755)
+ return
+
+ # Avoid shadowing packages with the same name when Python adds this directory to sys.path.
+ script_path = bin_dir / f"_{name}.py"
+ script_path.write_text(script, encoding="utf-8")
+ command = subprocess.list2cmdline([sys.executable, str(script_path)])
+ (bin_dir / f"{name}.cmd").write_text(
+ f"@echo off\n{command} %*\n",
+ encoding="utf-8",
+ )
+
+
+def _get_short_temp_dir() -> Optional[str]:
+ # Keep SSH control sockets below the Unix-domain socket path limit on macOS.
+ if IS_WINDOWS:
+ return None
+ return "/tmp" if Path("/tmp").is_dir() else None
+
+
+def _get_progress_script() -> str:
+ return f"""#!{sys.executable}
+import json
+import os
+from pathlib import Path
+import sys
+
+message = " ".join(sys.argv[1:]).strip()
+if not message and not sys.stdin.isatty():
+ message = sys.stdin.read().strip()
+if not message:
+ print("Usage: progress ", file=sys.stderr)
+ raise SystemExit(2)
+path = Path(os.environ.get("{_PROGRESS_ENV}", "{_PROGRESS_FILENAME}"))
+with path.open("a", encoding="utf-8") as f:
+ f.write(json.dumps({{"message": message}}, ensure_ascii=False) + "\\n")
+"""
+
+
+def _install_skills(workspace: Path) -> None:
+ source_dir = _get_skills_dir()
+ target_dir = workspace / ".claude" / "skills"
+ target_dir.mkdir(parents=True)
+ for skill_name in _SKILL_NAMES:
+ source = source_dir / skill_name
+ if not (source / "SKILL.md").is_file():
+ raise CLIError(f"Missing endpoint agent skill: {skill_name}")
+ shutil.copytree(source, target_dir / skill_name)
+
+
+def _get_skills_dir() -> Path:
+ source_path = Path(__file__).resolve()
+ candidates = (
+ source_path.parents[2] / "core" / "resources" / "endpoint_agent" / "skills",
+ source_path.parents[5] / "skills",
+ )
+ for candidate in candidates:
+ if (candidate / "dstack" / "SKILL.md").is_file():
+ return candidate
+ raise CLIError("Could not find packaged dstack skills")
+
+
+def _build_claude_command(*, auth: ClaudeAuth) -> list[str]:
+ command = [
+ auth.executable,
+ "-p",
+ "--output-format",
+ "stream-json",
+ "--verbose",
+ "--tools",
+ _CLAUDE_TOOLS,
+ "--allowedTools",
+ _CLAUDE_TOOLS,
+ "--disallowedTools",
+ "Task,NotebookEdit",
+ "--permission-mode",
+ "bypassPermissions",
+ "--model",
+ auth.model,
+ "--json-schema",
+ json.dumps(AGENT_FINAL_REPORT_JSON_SCHEMA),
+ ]
+ if auth.use_existing:
+ command[2:2] = ["--setting-sources", "project,local"]
+ else:
+ command[2:2] = ["--bare"]
+ if auth.effort is not None:
+ command[2:2] = ["--effort", auth.effort]
+ return command
+
+
+def _prepare_subprocess_command(command: list[str]) -> list[str]:
+ if not IS_WINDOWS or Path(command[0]).suffix.lower() not in {".bat", ".cmd"}:
+ return command
+ comspec = os.getenv("COMSPEC") or shutil.which("cmd.exe")
+ if comspec is None:
+ raise CLIError("Cannot run the Claude batch launcher because cmd.exe was not found")
+ return [comspec, "/d", "/s", "/c", subprocess.list2cmdline(command)]
+
+
+async def _read_process_stream(
+ *,
+ stream: asyncio.StreamReader,
+ path: Path,
+ parse_result: bool,
+ redacted_values: Sequence[str],
+) -> EndpointAgentProcessOutput:
+ output = EndpointAgentProcessOutput()
+ with path.open("a", encoding="utf-8") as f:
+ while True:
+ line = await stream.readline()
+ if not line:
+ return output
+ text = line.decode(errors="replace")
+ f.write(redact(text, redacted_values))
+ f.flush()
+ if not parse_result:
+ continue
+ try:
+ message = json.loads(text)
+ except json.JSONDecodeError:
+ continue
+ if not isinstance(message, dict) or message.get("type") != "result":
+ continue
+ if message.get("is_error"):
+ error = message.get("result") or "Claude failed"
+ output.error = redact(str(error), redacted_values)
+ structured_output = message.get("structured_output")
+ if isinstance(structured_output, dict):
+ output.report_data = structured_output
+ continue
+ result = message.get("result")
+ if isinstance(result, str):
+ try:
+ parsed = json.loads(result)
+ except json.JSONDecodeError:
+ continue
+ if isinstance(parsed, dict):
+ output.report_data = parsed
+
+
+async def _terminate_process(proc: asyncio.subprocess.Process) -> None:
+ if IS_WINDOWS:
+ await asyncio.to_thread(_terminate_windows_process_tree, proc.pid)
+ await proc.wait()
+ return
+ if hasattr(os, "killpg"):
+ with suppress(ProcessLookupError):
+ os.killpg(proc.pid, signal.SIGTERM)
+ else:
+ proc.terminate()
+ try:
+ await asyncio.wait_for(proc.wait(), timeout=3)
+ except asyncio.TimeoutError:
+ if hasattr(os, "killpg"):
+ with suppress(ProcessLookupError):
+ os.killpg(proc.pid, signal.SIGKILL)
+ else:
+ proc.kill()
+ await proc.wait()
+
+
+def _terminate_windows_process_tree(pid: int) -> None:
+ try:
+ root = psutil.Process(pid)
+ except psutil.NoSuchProcess:
+ return
+ processes = [*root.children(recursive=True), root]
+ for process in processes:
+ with suppress(psutil.NoSuchProcess):
+ process.terminate()
+ _, alive = psutil.wait_procs(processes, timeout=3)
+ for process in alive:
+ with suppress(psutil.NoSuchProcess):
+ process.kill()
+ psutil.wait_procs(alive, timeout=3)
+
+
+class _ProgressTailer:
+ def __init__(self, *, path: Path, redacted_values: Sequence[str]) -> None:
+ self._path = path
+ self._redacted_values = redacted_values
+ self._offset = 0
+
+ async def run(self) -> None:
+ while True:
+ self.flush()
+ await asyncio.sleep(1)
+
+ def flush(self) -> None:
+ if not self._path.exists():
+ return
+ with self._path.open("r", encoding="utf-8", errors="replace") as f:
+ f.seek(self._offset)
+ lines = f.readlines()
+ self._offset = f.tell()
+ for line in lines:
+ message = _parse_progress(line)
+ if message is not None:
+ print_endpoint_progress(redact(message, self._redacted_values))
+
+
+def _parse_progress(line: str) -> Optional[str]:
+ try:
+ value = json.loads(line)
+ except json.JSONDecodeError:
+ return line.strip() or None
+ if isinstance(value, str):
+ return value.strip() or None
+ if isinstance(value, dict) and isinstance(value.get("message"), str):
+ return value["message"].strip() or None
+ return None
diff --git a/src/dstack/_internal/cli/services/endpoint_preset_apply.py b/src/dstack/_internal/cli/services/endpoint_preset_apply.py
new file mode 100644
index 0000000000..1b0d8c97d7
--- /dev/null
+++ b/src/dstack/_internal/cli/services/endpoint_preset_apply.py
@@ -0,0 +1,145 @@
+import argparse
+from dataclasses import dataclass
+from typing import Optional
+
+from rich.table import Table
+
+from dstack._internal.cli.services.configurators.run import (
+ PreparedRunConfiguration,
+ ServiceConfigurator,
+)
+from dstack._internal.cli.services.endpoint_presets import EndpointPresetStore
+from dstack._internal.cli.utils.common import console
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.endpoint_presets import EndpointPresetRecipe
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.profiles import ProfileParams
+from dstack._internal.core.models.runs import RunPlan
+from dstack.api import Client
+
+
+@dataclass(frozen=True)
+class _PresetPlan:
+ recipe: EndpointPresetRecipe
+ prepared: PreparedRunConfiguration[ServiceConfiguration]
+
+
+def apply_endpoint_preset(
+ *,
+ api: Client,
+ configuration: EndpointConfiguration,
+ configuration_path: str,
+ recipe_id: Optional[str],
+ command_args: argparse.Namespace,
+ store: EndpointPresetStore,
+) -> None:
+ recipes = _get_matching_recipes(
+ store.list(),
+ configuration=configuration,
+ recipe_id=recipe_id,
+ )
+ if not recipes:
+ qualifier = f" recipe {recipe_id!r}" if recipe_id else ""
+ raise CLIError(
+ f"No matching endpoint preset{qualifier} for {configuration.model.api_model_name}"
+ )
+
+ configurator = ServiceConfigurator(api_client=api)
+ service_args = configurator.get_parser().parse_args([])
+ selected = _select_plan(
+ configuration=configuration,
+ configuration_path=configuration_path,
+ recipes=recipes,
+ configurator=configurator,
+ service_args=service_args,
+ )
+ _print_selected_recipe(selected.recipe)
+ configurator.apply_prepared_configuration(
+ prepared=selected.prepared,
+ command_args=command_args,
+ configurator_args=service_args,
+ )
+
+
+def _get_matching_recipes(
+ recipes: list[EndpointPresetRecipe],
+ *,
+ configuration: EndpointConfiguration,
+ recipe_id: Optional[str],
+) -> list[EndpointPresetRecipe]:
+ model_name = configuration.model.api_model_name
+ matches = []
+ for recipe in recipes:
+ service_model = recipe.service.model
+ if recipe_id is not None and recipe.id != recipe_id:
+ continue
+ if service_model is None or service_model.name.lower() != model_name.lower():
+ continue
+ if configuration.context_length is not None:
+ if recipe.context_length < configuration.context_length:
+ continue
+ if configuration.model.allows_variant_selection:
+ if recipe.base.lower() != model_name.lower():
+ continue
+ elif recipe.model != configuration.model.exact_repo:
+ continue
+ matches.append(recipe)
+ return matches
+
+
+def _select_plan(
+ *,
+ configuration: EndpointConfiguration,
+ configuration_path: str,
+ recipes: list[EndpointPresetRecipe],
+ configurator: ServiceConfigurator,
+ service_args: argparse.Namespace,
+) -> _PresetPlan:
+ first_plan: Optional[_PresetPlan] = None
+ for recipe in recipes:
+ service = _build_service(configuration, recipe)
+ prepared = configurator.prepare_configuration(
+ conf=service,
+ configuration_path=configuration_path,
+ configurator_args=service_args,
+ )
+ plan = _PresetPlan(recipe=recipe, prepared=prepared)
+ if first_plan is None:
+ first_plan = plan
+ if _has_available_offers(prepared.run_plan):
+ return plan
+ assert first_plan is not None
+ return first_plan
+
+
+def _build_service(
+ configuration: EndpointConfiguration,
+ recipe: EndpointPresetRecipe,
+) -> ServiceConfiguration:
+ service = recipe.service.copy(deep=True)
+ service.name = configuration.name
+ service.gateway = configuration.gateway
+ service.env.update(configuration.env)
+ for field in ProfileParams.__fields__:
+ value = getattr(configuration, field)
+ if value is not None:
+ setattr(service, field, value)
+ return service
+
+
+def _has_available_offers(plan: RunPlan) -> bool:
+ return bool(plan.job_plans) and all(
+ any(offer.availability.is_available() for offer in job_plan.offers)
+ for job_plan in plan.job_plans
+ )
+
+
+def _print_selected_recipe(recipe: EndpointPresetRecipe) -> None:
+ table = Table(box=None, show_header=False)
+ table.add_column(no_wrap=True)
+ table.add_column()
+ table.add_row("[bold]Preset[/]", recipe.base)
+ table.add_row("[bold]Recipe[/]", recipe.id)
+ console.print(table)
+ console.print()
diff --git a/src/dstack/_internal/cli/services/endpoint_preset_create.py b/src/dstack/_internal/cli/services/endpoint_preset_create.py
new file mode 100644
index 0000000000..a274c994a3
--- /dev/null
+++ b/src/dstack/_internal/cli/services/endpoint_preset_create.py
@@ -0,0 +1,285 @@
+import asyncio
+import json
+import secrets
+import uuid
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Optional, Sequence
+
+from dstack._internal.cli.services.endpoint_agent_runtime import (
+ EndpointAgentWorkspace,
+ build_endpoint_agent_env,
+ contains_redacted_value,
+ endpoint_agent_workspace,
+ get_claude_auth,
+ get_redacted_values,
+ get_sensitive_inherited_env_values,
+ print_endpoint_progress,
+ run_endpoint_agent,
+)
+from dstack._internal.cli.services.endpoint_preset_verify import (
+ build_verified_endpoint_preset,
+ load_endpoint_agent_report,
+ load_endpoint_benchmarks,
+)
+from dstack._internal.cli.services.endpoint_presets import EndpointPresetStore
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.endpoint_agent import AgentFinalReport
+from dstack._internal.core.models.endpoint_presets import EndpointPresetRecipe
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.fleets import FleetStatus
+from dstack._internal.core.services.endpoint_agent import (
+ format_endpoint_constraints,
+ get_endpoint_agent_system_prompt,
+)
+from dstack._internal.core.services.endpoint_presets import endpoint_preset_recipe_to_data
+from dstack.api import Client
+
+_RUN_STOP_TIMEOUT_SECONDS = 10 * 60
+
+
+@dataclass(frozen=True)
+class EndpointPresetCreateResult:
+ recipe: EndpointPresetRecipe
+ path: Path
+ final_run_id: uuid.UUID
+ final_run_name: str
+
+
+def create_endpoint_preset(
+ *,
+ api: Client,
+ configuration: EndpointConfiguration,
+ store: EndpointPresetStore,
+ keep_service: bool = False,
+ build_name: Optional[str] = None,
+) -> EndpointPresetCreateResult:
+ return asyncio.run(
+ _create_endpoint_preset(
+ api=api,
+ configuration=configuration,
+ store=store,
+ keep_service=keep_service,
+ build_name=build_name,
+ )
+ )
+
+
+async def _create_endpoint_preset(
+ *,
+ api: Client,
+ configuration: EndpointConfiguration,
+ store: EndpointPresetStore,
+ keep_service: bool = False,
+ build_name: Optional[str] = None,
+) -> EndpointPresetCreateResult:
+ build_name = build_name or _get_build_name(configuration.name)
+ allowed_fleets = _get_allowed_fleets(api, configuration)
+ if not allowed_fleets:
+ raise CLIError("The project has no active fleets available for preset creation")
+ auth = get_claude_auth()
+
+ endpoint_env = configuration.env.as_dict()
+ token = getattr(api.client, "_token", None)
+ if not isinstance(token, str) or not token:
+ raise CLIError("The configured dstack client has no authentication token")
+ redacted_values = get_redacted_values(
+ [
+ token,
+ auth.api_key or "",
+ *endpoint_env.values(),
+ *get_sensitive_inherited_env_values(),
+ ]
+ )
+ report: Optional[AgentFinalReport] = None
+ recipe: Optional[EndpointPresetRecipe] = None
+ preset_path: Optional[Path] = None
+ creation_succeeded = False
+ cleanup_error: Optional[str] = None
+ with endpoint_agent_workspace() as workspace:
+ env = build_endpoint_agent_env(
+ api=api,
+ endpoint_env=endpoint_env,
+ auth=auth,
+ workspace=workspace,
+ token=token,
+ )
+ prompt = _build_prompt(
+ configuration=configuration,
+ build_name=build_name,
+ allowed_fleets=allowed_fleets,
+ )
+ print_endpoint_progress(
+ f"Starting endpoint preset creation for {configuration.model.api_model_name}. "
+ f"Allowed fleets: {', '.join(allowed_fleets)}."
+ )
+ try:
+ process_output = await run_endpoint_agent(
+ prompt=prompt,
+ env=env,
+ workspace=workspace,
+ auth=auth,
+ redacted_values=redacted_values,
+ )
+ report = load_endpoint_agent_report(
+ output=process_output,
+ workspace=workspace,
+ redacted_values=redacted_values,
+ )
+ benchmarks = load_endpoint_benchmarks(workspace, report=report)
+ run = api.client.runs.get(api.project, report.run_name)
+ recipe = build_verified_endpoint_preset(
+ run=run,
+ endpoint_configuration=configuration,
+ report=report,
+ benchmarks=benchmarks,
+ )
+ if contains_redacted_value(endpoint_preset_recipe_to_data(recipe), redacted_values):
+ raise CLIError("Generated endpoint preset contains a secret value")
+ preset_path = store.save(recipe)
+ print_endpoint_progress(
+ f"Saved endpoint preset recipe {recipe.id} for {recipe.base} at {preset_path}."
+ )
+ creation_succeeded = True
+ finally:
+ keep_final_service = keep_service and creation_succeeded
+ try:
+ await _cleanup_runs(
+ api=api,
+ build_name=build_name,
+ workspace=workspace,
+ final_run_name=report.run_name if report is not None else None,
+ keep_final_service=keep_final_service,
+ )
+ except Exception as e:
+ cleanup_error = str(e)
+ if keep_final_service:
+ with suppress(Exception):
+ await _cleanup_runs(
+ api=api,
+ build_name=build_name,
+ workspace=workspace,
+ final_run_name=report.run_name if report is not None else None,
+ )
+
+ if cleanup_error is not None:
+ raise CLIError(f"Failed to clean up preset creation runs: {cleanup_error}")
+ assert recipe is not None
+ assert preset_path is not None
+ assert report is not None
+ assert report.run_id is not None
+ assert report.run_name is not None
+ return EndpointPresetCreateResult(
+ recipe=recipe,
+ path=preset_path,
+ final_run_id=report.run_id,
+ final_run_name=report.run_name,
+ )
+
+
+def _get_build_name(endpoint_name: Optional[str]) -> str:
+ if endpoint_name is None:
+ raise CLIError("Endpoint name is required. Set `name` in the configuration or use --name")
+ suffix = secrets.token_hex(3)
+ # Leave room for the numeric submission suffix while retaining a recognizable prefix.
+ prefix = endpoint_name[:28].rstrip("-")
+ return f"{prefix}-{suffix}"
+
+
+def _get_allowed_fleets(api: Client, configuration: EndpointConfiguration) -> tuple[str, ...]:
+ if configuration.fleets is not None:
+ return tuple(
+ fleet.format() if hasattr(fleet, "format") else str(fleet)
+ for fleet in configuration.fleets
+ )
+ fleets = api.client.fleets.list(api.project, include_imported=True)
+ return tuple(
+ fleet.name if fleet.project_name == api.project else f"{fleet.project_name}/{fleet.name}"
+ for fleet in fleets
+ if fleet.status == FleetStatus.ACTIVE
+ )
+
+
+def _build_prompt(
+ *,
+ configuration: EndpointConfiguration,
+ build_name: str,
+ allowed_fleets: Sequence[str],
+) -> str:
+ context_lines = [f"- service_model_name: {configuration.model.api_model_name}"]
+ if configuration.model.allows_variant_selection:
+ context_lines.append(f"- base_model: {configuration.model.api_model_name}")
+ else:
+ context_lines.append(f"- model_repo: {configuration.model.exact_repo}")
+ if configuration.context_length is not None:
+ context_lines.append(f"- context_length: {configuration.context_length}")
+ return f"""{get_endpoint_agent_system_prompt()}
+
+Endpoint context:
+- endpoint_name: {build_name}
+{chr(10).join(context_lines)}
+
+{
+ format_endpoint_constraints(
+ configuration,
+ configuration.env.as_dict(),
+ allowed_fleets=allowed_fleets,
+ )
+ }
+"""
+
+
+async def _cleanup_runs(
+ *,
+ api: Client,
+ build_name: str,
+ workspace: EndpointAgentWorkspace,
+ final_run_name: Optional[str],
+ keep_final_service: bool = False,
+) -> None:
+ run_names = _load_submitted_run_names(workspace.submissions_path)
+ if final_run_name is not None:
+ run_names.append(final_run_name)
+ run_names = list(dict.fromkeys(run_names))
+ expected_prefix = f"{build_name}-"
+ run_names = [name for name in run_names if name.startswith(expected_prefix)]
+ if keep_final_service:
+ run_names = [name for name in run_names if name != final_run_name]
+ active_names = []
+ for name in run_names:
+ run = api.runs.get(name)
+ if run is not None and not run.status.is_finished():
+ active_names.append(name)
+ if not active_names:
+ return
+ print_endpoint_progress(f"Stopping preset creation runs: {', '.join(active_names)}.")
+ api.client.runs.stop(api.project, active_names, abort=False)
+ deadline = asyncio.get_running_loop().time() + _RUN_STOP_TIMEOUT_SECONDS
+ pending = set(active_names)
+ while pending:
+ if asyncio.get_running_loop().time() >= deadline:
+ raise CLIError(f"Timed out waiting for runs to stop: {', '.join(sorted(pending))}")
+ for name in list(pending):
+ run = api.runs.get(name)
+ if run is None or run.status.is_finished():
+ pending.remove(name)
+ if pending:
+ await asyncio.sleep(2)
+ print_endpoint_progress("All preset creation runs stopped.")
+
+
+def _load_submitted_run_names(path: Path) -> list[str]:
+ if not path.exists():
+ return []
+ names = []
+ for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+ try:
+ value = json.loads(line)
+ except json.JSONDecodeError:
+ continue
+ if isinstance(value, dict) and isinstance(value.get("name"), str):
+ name = value["name"].strip()
+ if name:
+ names.append(name)
+ return names
diff --git a/src/dstack/_internal/cli/services/endpoint_preset_verify.py b/src/dstack/_internal/cli/services/endpoint_preset_verify.py
new file mode 100644
index 0000000000..538106961c
--- /dev/null
+++ b/src/dstack/_internal/cli/services/endpoint_preset_verify.py
@@ -0,0 +1,182 @@
+import json
+from pathlib import Path
+from typing import Any, Optional, Sequence
+from urllib.parse import urlparse
+
+from pydantic import ValidationError
+
+from dstack._internal.cli.services.endpoint_agent_runtime import (
+ EndpointAgentProcessOutput,
+ EndpointAgentWorkspace,
+ redact,
+)
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.endpoint_agent import AgentFinalReport
+from dstack._internal.core.models.endpoint_presets import (
+ EndpointBenchmark,
+ EndpointBenchmarkClient,
+ EndpointBenchmarkTarget,
+ EndpointPresetRecipe,
+ EndpointPresetValidationReplica,
+)
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.envs import EnvSentinel
+from dstack._internal.core.models.runs import JobStatus, Run, RunStatus
+from dstack._internal.core.services.endpoint_presets import (
+ build_endpoint_preset_recipe,
+ resources_spec_from_instance_resources,
+)
+
+
+def load_endpoint_agent_report(
+ *,
+ output: EndpointAgentProcessOutput,
+ workspace: EndpointAgentWorkspace,
+ redacted_values: Sequence[str],
+) -> AgentFinalReport:
+ report_data = output.report_data or _load_json_object(workspace.final_report_path)
+ if report_data is None:
+ raise CLIError(
+ redact(
+ output.error or "Claude exited without a final report",
+ redacted_values,
+ )
+ )
+ try:
+ report = AgentFinalReport.parse_obj(report_data)
+ except ValidationError as e:
+ raise CLIError(f"Claude returned an invalid final report: {e}") from e
+ if not report.success:
+ raise CLIError(
+ redact(
+ report.failure_summary or "Claude did not create a preset",
+ redacted_values,
+ )
+ )
+ return report
+
+
+def load_endpoint_benchmarks(
+ workspace: EndpointAgentWorkspace,
+ *,
+ report: AgentFinalReport,
+) -> list[EndpointBenchmark]:
+ path = workspace.benchmarks_path
+ if not path.exists():
+ raise CLIError(f"{path.name} is missing")
+ benchmarks: list[EndpointBenchmark] = []
+ for line_num, line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
+ if not line.strip():
+ continue
+ try:
+ benchmark = EndpointBenchmark.parse_raw(line)
+ except (ValidationError, ValueError) as e:
+ raise CLIError(f"{path.name} line {line_num} is invalid") from e
+ if benchmark.run_name != report.run_name:
+ continue
+ if benchmark.run_id is not None and benchmark.run_id != report.run_id:
+ continue
+ benchmarks.append(benchmark)
+ if not benchmarks:
+ raise CLIError("Claude did not benchmark the final service run")
+ if any(not benchmark.success for benchmark in benchmarks):
+ raise CLIError("Claude reported a failed benchmark for the final service run")
+ return benchmarks
+
+
+def build_verified_endpoint_preset(
+ *,
+ run: Run,
+ endpoint_configuration: EndpointConfiguration,
+ report: AgentFinalReport,
+ benchmarks: list[EndpointBenchmark],
+) -> EndpointPresetRecipe:
+ if run.id != report.run_id or run.run_spec.run_name != report.run_name:
+ raise CLIError("Claude final report identifies a different service run")
+ if run.status != RunStatus.RUNNING or run.service is None:
+ raise CLIError("Claude final service is not running")
+ service = run.run_spec.configuration
+ if not isinstance(service, ServiceConfiguration) or service.model is None:
+ raise CLIError("Claude final run is not a model service")
+ if service.model.name != endpoint_configuration.model.api_model_name:
+ raise CLIError("Claude final service model name does not match the endpoint request")
+ assert report.base is not None
+ assert report.model is not None
+ assert report.context_length is not None
+ if endpoint_configuration.model.allows_variant_selection:
+ if report.base != endpoint_configuration.model.api_model_name:
+ raise CLIError("Claude final report base does not match the endpoint request")
+ elif report.model != endpoint_configuration.model.exact_repo:
+ raise CLIError("Claude changed an exact model request")
+ if (
+ endpoint_configuration.context_length is not None
+ and report.context_length < endpoint_configuration.context_length
+ ):
+ raise CLIError("Claude final service does not meet the requested context length")
+
+ target_type = (
+ "gateway" if urlparse(run.service.url).scheme in {"http", "https"} else "server-proxy"
+ )
+ benchmarks = [
+ benchmark.copy(
+ update={
+ "target": EndpointBenchmarkTarget(type=target_type),
+ "client": EndpointBenchmarkClient(type="local"),
+ }
+ )
+ for benchmark in benchmarks
+ ]
+ portable_service = service.copy(deep=True)
+ # The CLI resolved endpoint env references before submission; presets retain the references.
+ for key in endpoint_configuration.env:
+ if key in portable_service.env:
+ portable_service.env[key] = EnvSentinel(key=key)
+ return build_endpoint_preset_recipe(
+ service=portable_service,
+ validation_replicas=_get_validation_replicas(run, service),
+ base_model=report.base,
+ recipe_model=report.model,
+ context_length=report.context_length,
+ benchmarks=benchmarks,
+ )
+
+
+def _get_validation_replicas(
+ run: Run,
+ service: ServiceConfiguration,
+) -> list[EndpointPresetValidationReplica]:
+ replicas: list[EndpointPresetValidationReplica] = []
+ for group in service.replica_groups:
+ resources = []
+ for job in sorted(run.jobs, key=lambda job: job.job_spec.replica_num):
+ if job.job_spec.job_num != 0 or job.job_spec.replica_group != group.name:
+ continue
+ submissions = [
+ submission
+ for submission in job.job_submissions
+ if submission.deployment_num == run.deployment_num
+ and submission.status == JobStatus.RUNNING
+ ]
+ if not submissions:
+ continue
+ runtime_data = submissions[-1].job_runtime_data
+ if runtime_data is None or runtime_data.offer is None:
+ raise CLIError("Final service run does not expose actual instance resources")
+ resources.append(
+ resources_spec_from_instance_resources(runtime_data.offer.instance.resources)
+ )
+ if not resources:
+ raise CLIError(f"Final service replica group {group.name!r} has no running replicas")
+ replicas.append(EndpointPresetValidationReplica(resources=resources))
+ return replicas
+
+
+def _load_json_object(path: Path) -> Optional[dict[str, Any]]:
+ if not path.exists():
+ return None
+ try:
+ value = json.loads(path.read_text(encoding="utf-8"))
+ except json.JSONDecodeError:
+ return None
+ return value if isinstance(value, dict) else None
diff --git a/src/dstack/_internal/cli/services/endpoint_presets.py b/src/dstack/_internal/cli/services/endpoint_presets.py
new file mode 100644
index 0000000000..a2af40b410
--- /dev/null
+++ b/src/dstack/_internal/cli/services/endpoint_presets.py
@@ -0,0 +1,136 @@
+import os
+import sys
+import tempfile
+from pathlib import Path
+from typing import List, TextIO
+
+import yaml
+from pydantic import ValidationError
+
+from dstack._internal.core.errors import CLIError, ConfigurationError
+from dstack._internal.core.models.endpoint_presets import EndpointPresetRecipe
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.services.endpoint_presets import endpoint_preset_recipe_to_data
+from dstack._internal.utils.common import get_dstack_dir
+
+
+class EndpointPresetStore:
+ def __init__(self, root: Path | None = None) -> None:
+ self.root = root or get_dstack_dir() / "presets"
+
+ def list(self) -> list[EndpointPresetRecipe]:
+ if not self.root.exists():
+ return []
+ recipes = [self._load(path) for path in self.root.glob("models--*/*.yaml")]
+ return sorted(recipes, key=lambda recipe: (recipe.base.lower(), recipe.id))
+
+ def get(self, recipe_id: str) -> EndpointPresetRecipe | None:
+ paths = self._find_recipe_paths(recipe_id)
+ if not paths:
+ return None
+ if len(paths) > 1:
+ raise CLIError(f"Endpoint preset recipe ID {recipe_id!r} is not unique")
+ path = paths[0]
+ recipe = self._load(path)
+ if recipe.id != recipe_id:
+ raise CLIError(f"Endpoint preset file {path} does not match its path")
+ return recipe
+
+ def save(self, recipe: EndpointPresetRecipe) -> Path:
+ path = self._path(recipe.base, recipe.id)
+ path.parent.mkdir(parents=True, exist_ok=True)
+ content = yaml.safe_dump(endpoint_preset_recipe_to_data(recipe), sort_keys=False)
+ fd, temporary_path = tempfile.mkstemp(
+ dir=path.parent,
+ prefix=f".{recipe.id}.",
+ suffix=".tmp",
+ )
+ try:
+ with os.fdopen(fd, "w", encoding="utf-8") as f:
+ f.write(content)
+ f.flush()
+ os.fsync(f.fileno())
+ os.replace(temporary_path, path)
+ finally:
+ try:
+ Path(temporary_path).unlink()
+ except FileNotFoundError:
+ pass
+ return path
+
+ def delete_recipe(self, recipe_id: str) -> bool:
+ recipe = self.get(recipe_id)
+ if recipe is None:
+ return False
+ path = self._path(recipe.base, recipe.id)
+ path.unlink()
+ try:
+ path.parent.rmdir()
+ except OSError:
+ pass
+ return True
+
+ def delete_preset(self, base: str) -> int:
+ directory = self._directory(base)
+ paths = list(directory.glob("*.yaml"))
+ recipes = [self._load(path) for path in paths]
+ if any(recipe.base != base for recipe in recipes):
+ raise CLIError(f"Endpoint preset directory {directory} contains another base model")
+ for path in paths:
+ path.unlink()
+ try:
+ directory.rmdir()
+ except OSError:
+ pass
+ return len(recipes)
+
+ def _load(self, path: Path) -> EndpointPresetRecipe:
+ try:
+ with path.open(encoding="utf-8") as f:
+ return EndpointPresetRecipe.parse_obj(yaml.safe_load(f))
+ except (OSError, ValidationError, yaml.YAMLError) as e:
+ raise CLIError(f"Invalid endpoint preset file {path}: {e}") from e
+
+ def _path(self, base: str, recipe_id: str) -> Path:
+ if not recipe_id or any(char in recipe_id for char in "/\\"):
+ raise CLIError("Endpoint preset recipe ID must not contain path separators")
+ return self._directory(base) / f"{recipe_id}.yaml"
+
+ def _find_recipe_paths(self, recipe_id: str) -> List[Path]:
+ if not recipe_id or any(char in recipe_id for char in "/\\"):
+ raise CLIError("Endpoint preset recipe ID must not contain path separators")
+ return [
+ path
+ for directory in self.root.glob("models--*")
+ if (path := directory / f"{recipe_id}.yaml").is_file()
+ ]
+
+ def _directory(self, base: str) -> Path:
+ directory = "models--" + base.replace("/", "--").replace("\\", "--")
+ return self.root / directory
+
+
+def load_endpoint_configuration(path: str) -> tuple[str, EndpointConfiguration]:
+ if path == "-":
+ return "-", _parse_endpoint_configuration(sys.stdin)
+ configuration_path = Path(path)
+ if not configuration_path.is_file():
+ raise ConfigurationError(f"Configuration file {path} does not exist")
+ try:
+ with configuration_path.open(encoding="utf-8") as f:
+ configuration = _parse_endpoint_configuration(f)
+ except OSError as e:
+ raise ConfigurationError(f"Failed to load configuration from {path}") from e
+ return str(configuration_path.resolve()), configuration
+
+
+def _parse_endpoint_configuration(stream: TextIO) -> EndpointConfiguration:
+ try:
+ data = yaml.safe_load(stream)
+ if not isinstance(data, dict):
+ raise ConfigurationError("Endpoint configuration must be a YAML object")
+ return EndpointConfiguration.parse_obj(data)
+ except ValidationError as e:
+ raise ConfigurationError(e) from e
+ except yaml.YAMLError as e:
+ raise ConfigurationError(f"Invalid endpoint configuration: {e}") from e
diff --git a/src/dstack/_internal/cli/utils/endpoint_presets.py b/src/dstack/_internal/cli/utils/endpoint_presets.py
new file mode 100644
index 0000000000..813ba2754b
--- /dev/null
+++ b/src/dstack/_internal/cli/utils/endpoint_presets.py
@@ -0,0 +1,76 @@
+from collections import defaultdict
+
+from rich.table import Table
+
+from dstack._internal.cli.utils.common import add_row_from_dict, console
+from dstack._internal.core.models.endpoint_presets import EndpointPresetRecipe
+from dstack._internal.utils.common import pretty_resources
+
+
+def print_endpoint_presets(recipes: list[EndpointPresetRecipe], verbose: bool = False) -> None:
+ table = Table(box=None)
+ table.add_column("MODEL", no_wrap=True)
+ table.add_column("RESOURCES" if verbose else "GPU")
+ recipes_by_base: dict[str, list[EndpointPresetRecipe]] = defaultdict(list)
+ for recipe in recipes:
+ recipes_by_base[recipe.base].append(recipe)
+
+ for base, base_recipes in recipes_by_base.items():
+ add_row_from_dict(table, {"MODEL": f"[bold]{base}[/]"})
+ for recipe in base_recipes:
+ _add_recipe(table, recipe, verbose=verbose)
+ console.print(table)
+ console.print()
+
+
+def _add_recipe(table: Table, recipe: EndpointPresetRecipe, *, verbose: bool) -> None:
+ groups = recipe.service.replica_groups
+ column = "RESOURCES" if verbose else "GPU"
+ add_row_from_dict(
+ table,
+ {
+ "MODEL": f"[secondary] recipe={recipe.id}[/]",
+ column: _format_resources(groups[0].resources, verbose=verbose),
+ },
+ )
+ if recipe.model != recipe.base:
+ add_row_from_dict(
+ table,
+ {"MODEL": f" repo={recipe.model}"},
+ style="secondary",
+ )
+ if len(groups) > 1:
+ for group in groups:
+ add_row_from_dict(
+ table,
+ {
+ "MODEL": f" group={group.name}",
+ column: _format_resources(group.resources, verbose=verbose),
+ },
+ style="secondary",
+ )
+ if verbose:
+ add_row_from_dict(
+ table,
+ {"MODEL": f" context_length={recipe.context_length}"},
+ style="secondary",
+ )
+
+
+def _format_resources(resources, *, verbose: bool) -> str:
+ if resources is None:
+ return "-"
+ if verbose:
+ return resources.pretty_format()
+ gpu = resources.gpu
+ if gpu is None or gpu.count.max == 0:
+ return "-"
+ formatted = pretty_resources(
+ gpu_vendor=gpu.vendor,
+ gpu_name=",".join(gpu.name) if gpu.name else None,
+ gpu_count=gpu.count,
+ gpu_memory=gpu.memory,
+ total_gpu_memory=gpu.total_memory,
+ compute_capability=gpu.compute_capability,
+ )
+ return formatted.removeprefix("gpu=") or "-"
diff --git a/src/dstack/_internal/core/models/endpoint_agent.py b/src/dstack/_internal/core/models/endpoint_agent.py
new file mode 100644
index 0000000000..aa70653063
--- /dev/null
+++ b/src/dstack/_internal/core/models/endpoint_agent.py
@@ -0,0 +1,44 @@
+import uuid
+from typing import Optional
+
+from pydantic import PositiveInt, root_validator
+
+from dstack._internal.core.models.common import CoreModel
+
+AGENT_FINAL_REPORT_JSON_SCHEMA = {
+ "type": "object",
+ "properties": {
+ "success": {"type": "boolean"},
+ "run_id": {"type": "string"},
+ "run_name": {"type": "string"},
+ "service_yaml": {"type": "string"},
+ "base": {"type": "string"},
+ "model": {"type": "string"},
+ "context_length": {"type": "integer", "minimum": 1},
+ "failure_summary": {"type": "string"},
+ },
+ "required": ["success"],
+ "additionalProperties": False,
+}
+
+
+class AgentFinalReport(CoreModel):
+ success: bool
+ run_id: Optional[uuid.UUID] = None
+ run_name: Optional[str] = None
+ service_yaml: Optional[str] = None
+ base: Optional[str] = None
+ model: Optional[str] = None
+ context_length: Optional[PositiveInt] = None
+ failure_summary: Optional[str] = None
+
+ @root_validator
+ def validate_report(cls, values: dict) -> dict:
+ if values.get("success"):
+ required = ("run_id", "run_name", "service_yaml", "base", "model", "context_length")
+ missing = [field for field in required if values.get(field) in (None, "")]
+ if missing:
+ raise ValueError("successful agent report must include " + ", ".join(missing))
+ elif not values.get("failure_summary"):
+ raise ValueError("failed agent report must include failure_summary")
+ return values
diff --git a/src/dstack/_internal/core/models/endpoint_presets.py b/src/dstack/_internal/core/models/endpoint_presets.py
new file mode 100644
index 0000000000..76c93cf483
--- /dev/null
+++ b/src/dstack/_internal/core/models/endpoint_presets.py
@@ -0,0 +1,172 @@
+import re
+import uuid
+from typing import Annotated, Any, Literal, Optional, Union
+
+from pydantic import (
+ Field,
+ PositiveFloat,
+ PositiveInt,
+ parse_obj_as,
+ root_validator,
+ validator,
+)
+
+from dstack._internal.core.models.common import CoreModel
+from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.profiles import ProfileParams
+from dstack._internal.core.models.resources import CPUSpec, ResourcesSpec
+
+
+class EndpointBenchmarkWorkload(CoreModel):
+ dataset_name: str
+ streaming: bool
+ max_concurrency: PositiveInt
+ request_rate: Optional[Union[Literal["inf"], PositiveFloat]] = None
+ num_prompts: PositiveInt
+ input_tokens: PositiveInt
+ output_tokens: PositiveInt
+ ignore_eos: Optional[bool] = None
+ random_range_ratio: Annotated[Optional[float], Field(ge=0, le=1)] = None
+
+ @validator("dataset_name")
+ def validate_dataset_name(cls, value: str) -> str:
+ if not value.strip():
+ raise ValueError("dataset_name must be non-empty")
+ return value
+
+
+class EndpointBenchmarkTarget(CoreModel):
+ type: Literal["gateway", "server-proxy"]
+
+
+class EndpointBenchmarkClient(CoreModel):
+ type: Literal["local"]
+
+
+class EndpointBenchmark(CoreModel):
+ success: bool
+ run_name: str
+ run_type: Literal["service"]
+ tool: str
+ command: str
+ workload: EndpointBenchmarkWorkload
+ metrics: Optional[dict[str, Any]] = None
+ failure_summary: Optional[str] = None
+ run_id: Optional[uuid.UUID] = None
+ tool_version: Optional[str] = None
+ target: Optional[EndpointBenchmarkTarget] = None
+ client: Optional[EndpointBenchmarkClient] = None
+
+ @validator("run_name", "tool", "command")
+ def validate_non_empty(cls, value: str) -> str:
+ if not value.strip():
+ raise ValueError("value must be non-empty")
+ return value
+
+ @validator("command")
+ def validate_command_has_no_bearer_token(cls, value: str) -> str:
+ for match in re.finditer(r"(?i)\bbearer\s+([^\s\"']+)", value):
+ token = match.group(1)
+ if token.startswith("$") or "redacted" in token.lower() or set(token) == {"*"}:
+ continue
+ raise ValueError("command must not contain a bearer token value")
+ return value
+
+ @root_validator
+ def validate_result(cls, values: dict) -> dict:
+ if values.get("success"):
+ if not values.get("metrics"):
+ raise ValueError("successful benchmark must include metrics")
+ elif not values.get("failure_summary"):
+ raise ValueError("failed benchmark must include failure_summary")
+ return values
+
+
+class EndpointPresetValidationReplica(CoreModel):
+ resources: list[ResourcesSpec]
+ """Exact resources for each running replica in this service replica group."""
+
+
+class EndpointPresetValidation(CoreModel):
+ replicas: list[EndpointPresetValidationReplica]
+ """Ordered to match `ServiceConfiguration.replica_groups`."""
+ benchmarks: list[EndpointBenchmark] = Field(default_factory=list)
+
+
+class EndpointPresetRecipe(CoreModel):
+ base: str
+ """Base model used for local preset lookup."""
+ id: str
+ model: str
+ """Exact repo/path loaded by the service command."""
+ context_length: PositiveInt
+ """Token context length this recipe was verified to support."""
+ service: ServiceConfiguration
+ validations: list[EndpointPresetValidation]
+
+ @validator("base", "id", "model")
+ def validate_non_empty(cls, value: str) -> str:
+ if not value.strip():
+ raise ValueError("value must be non-empty")
+ return value
+
+ @root_validator
+ def validate_recipe(cls, values: dict) -> dict:
+ service = values.get("service")
+ validations = values.get("validations")
+ if service is None or validations is None:
+ return values
+ if service.model is None:
+ raise ValueError("preset recipe service must specify model")
+ if any(group.resources is None for group in service.replica_groups):
+ raise ValueError("preset recipe service must specify resources")
+ if service.name is not None or service.gateway is not None:
+ raise ValueError("preset recipe service must not specify name or gateway")
+ if any(getattr(service, field) is not None for field in ProfileParams.__fields__):
+ raise ValueError("preset recipe service must not specify placement constraints")
+ if not validations:
+ raise ValueError("preset recipe must include validation evidence")
+ for validation in validations:
+ if len(validation.replicas) != len(service.replica_groups):
+ raise ValueError(
+ "preset validation replicas must match service replica group order"
+ )
+ if not validation.benchmarks or any(
+ not benchmark.success for benchmark in validation.benchmarks
+ ):
+ raise ValueError("preset validation must include a successful benchmark")
+ if any(
+ benchmark.target is None or benchmark.client is None
+ for benchmark in validation.benchmarks
+ ):
+ raise ValueError("preset benchmark must specify target and client")
+ for replica_group in validation.replicas:
+ if not replica_group.resources:
+ raise ValueError("preset validation replicas must specify resources")
+ for resources in replica_group.resources:
+ _validate_exact_resources(resources)
+ return values
+
+
+def _validate_exact_resources(resources: ResourcesSpec) -> None:
+ cpu = parse_obj_as(CPUSpec, resources.cpu)
+ if not _is_exact(cpu.count) or not _is_exact(resources.memory):
+ raise ValueError("preset validation resources must be exact")
+ if resources.disk is None or not _is_exact(resources.disk.size):
+ raise ValueError("preset validation resources must be exact")
+ gpu = resources.gpu
+ if gpu is None or not _is_exact(gpu.count):
+ raise ValueError("preset validation resources must be exact")
+ if gpu.count.min == 0:
+ return
+ if gpu.name is None or len(gpu.name) != 1 or not _is_exact(gpu.memory):
+ raise ValueError("preset validation resources must be exact")
+
+
+def _is_exact(value) -> bool:
+ return (
+ value is not None
+ and value.min is not None
+ and value.max is not None
+ and value.min == value.max
+ )
diff --git a/src/dstack/_internal/core/models/endpoints.py b/src/dstack/_internal/core/models/endpoints.py
new file mode 100644
index 0000000000..ee39a8e6cd
--- /dev/null
+++ b/src/dstack/_internal/core/models/endpoints.py
@@ -0,0 +1,98 @@
+from typing import Annotated, Any, Literal, Optional, Union
+
+from pydantic import Field, PositiveInt, validator
+
+from dstack._internal.core.models.common import CoreModel, EntityReference
+from dstack._internal.core.models.envs import Env
+from dstack._internal.core.models.profiles import ProfileParams
+
+
+class EndpointModelRepo(CoreModel):
+ repo: str
+ """Exact repo/path to deploy."""
+ name: Optional[str] = None
+ """Client-facing model name. Defaults to `repo`."""
+
+ @property
+ def api_model_name(self) -> str:
+ return self.name or self.repo
+
+ @property
+ def exact_repo(self) -> str:
+ return self.repo
+
+ @property
+ def allows_variant_selection(self) -> bool:
+ return False
+
+ @validator("repo")
+ def validate_repo(cls, value: str) -> str:
+ return _validate_model(value, field="repo")
+
+ @validator("name")
+ def validate_name(cls, value: Optional[str]) -> Optional[str]:
+ if value is None:
+ return None
+ return _validate_model(value, field="name")
+
+
+class EndpointModelBase(CoreModel):
+ base: str
+ """Base model for which the agent may select a compatible variant."""
+
+ @property
+ def api_model_name(self) -> str:
+ return self.base
+
+ @property
+ def exact_repo(self) -> None:
+ return None
+
+ @property
+ def allows_variant_selection(self) -> bool:
+ return True
+
+ @validator("base")
+ def validate_base(cls, value: str) -> str:
+ return _validate_model(value, field="base")
+
+
+EndpointModelSpec = Union[EndpointModelRepo, EndpointModelBase]
+
+
+class EndpointConfiguration(ProfileParams):
+ type: Literal["endpoint"] = "endpoint"
+ name: Optional[str] = None
+ model: Annotated[
+ EndpointModelSpec,
+ Field(
+ description=(
+ "The model to serve. Use a string or `repo` for an exact repo/path, "
+ "or `base` to allow compatible model variants."
+ )
+ ),
+ ]
+ context_length: Optional[PositiveInt] = None
+ """Minimum context length required from the endpoint."""
+ recipe: Optional[str] = None
+ """Recipe ID to use when applying an endpoint preset."""
+ gateway: Optional[Union[bool, EntityReference, str]] = None
+ env: Env = Env()
+
+ @validator("model", pre=True)
+ def parse_model(cls, value: Any) -> Any:
+ if isinstance(value, str):
+ return {"repo": _validate_model(value, field="model")}
+ return value
+
+ @validator("recipe")
+ def validate_recipe(cls, value: Optional[str]) -> Optional[str]:
+ if value is not None and not value.strip():
+ raise ValueError("Endpoint recipe must be a non-empty string")
+ return value
+
+
+def _validate_model(value: Any, *, field: str) -> str:
+ if not isinstance(value, str) or not value.strip():
+ raise ValueError(f"Endpoint model {field} must be a non-empty string")
+ return value
diff --git a/src/dstack/_internal/core/resources/endpoint_agent/system_prompt.md b/src/dstack/_internal/core/resources/endpoint_agent/system_prompt.md
new file mode 100644
index 0000000000..0d0a9042c2
--- /dev/null
+++ b/src/dstack/_internal/core/resources/endpoint_agent/system_prompt.md
@@ -0,0 +1,312 @@
+# Objective
+
+You are the endpoint preset creation agent for dstack. Produce one final dstack
+service that can be saved as a reusable preset recipe. Report success only after
+that service answers a real request using `service_model_name` from `Endpoint
+context:` through the dstack service URL.
+
+# Requested Model
+
+The `Endpoint context:` block contains either `model_repo` or `base_model`.
+
+- If it contains `model_repo`, deploy that repo/path exactly.
+- If it contains `base_model`, choose a repo/path compatible with `base_model`
+ that best fits performance and hardware within the endpoint constraints,
+ allowed fleets, backends, and offers. A variant can be the base repo itself, a
+ different precision or quantization, or another trusted repo compatible with
+ `base_model`.
+
+If `Endpoint context:` contains `context_length`, the selected repo/path and
+final service must support at least that context length.
+
+Use the real `dstack` CLI and shell commands in this workspace. Load and follow
+`/dstack` for dstack CLI/YAML syntax. Load and follow `/dstack-prototyping` for
+how to test a model-serving recipe with tasks before verifying it as a service.
+The skill files are installed under `.claude/skills` if you need to inspect
+them directly.
+
+# Progress
+
+Write endpoint progress with:
+
+```bash
+progress "message"
+```
+
+The helper appends to `progress.jsonl`; the caller shows only the message text.
+
+Write progress before your first investigation action, whenever you submit a
+run, when new evidence changes the plan, when model verification succeeds or
+fails, and before the final report. Messages should explain what you tried, what
+happened, and what you will do next. Do not put raw YAML, command output, long
+tables, traces, or secrets in progress.
+
+Write a short progress message for every meaningful choice or action. Each
+message should say what you did or chose, why, what evidence you used, what
+happened, and what you will do next. Include the run name when a run is
+involved. Include the fleet or backend when a fleet or backend is involved. Do
+not use generic phase labels as the message.
+
+# Workspace Files
+
+Create local files only in the current workspace.
+
+`submissions.jsonl` is append-only. For every dstack task or service you submit,
+append one JSON line when you submit it and more JSON lines when you learn its
+run id, status, URL, or final outcome.
+
+Use these fields:
+
+- `event`: `submit`, `update`, or `final`
+- `name`: submitted dstack run name
+- `type`: `task` or `service`
+- `status`: current known status
+- `config_path`: YAML file path, when applicable
+- `run_id`: run id, when known
+- `reason`: why this run exists or why its status changed
+- `service_url`: service URL, when known
+
+Example:
+
+```json
+{"event":"submit","name":"qwen-endpoint-1","type":"task","status":"submitted","config_path":"qwen-endpoint-1.dstack.yml","reason":"test the serving image and local model request on an allowed fleet"}
+{"event":"update","name":"qwen-endpoint-1","type":"task","status":"running","run_id":"..."}
+```
+
+Do not rewrite previous `submissions.jsonl` lines; the latest line for a run is
+the current record. Stop runs you no longer need unless they are still needed for
+attach/SSH debugging, logs, or backend diagnosis.
+
+After stopping a task or service, follow `/dstack` structured status guidance
+and confirm that the run reached a terminal status before continuing.
+
+# Run Names
+
+Do not use the endpoint name itself as a submitted run name.
+
+Use only `-` for dstack runs submitted for
+this endpoint. The first submitted run is `-1`, the second is
+`-2`, and so on. Do not add framework, hardware, role, or purpose
+suffixes to run names; record those details in workspace files and progress.
+
+# Endpoint Constraints
+
+Use existing allowed fleets only. Do not create, delete, apply, or edit fleets,
+including `nodes`, `target`, `idle_duration`, backends, resources, max nodes, or
+ownership.
+
+If the endpoint config lists fleets, use only those fleets. Otherwise, use the
+existing project/imported fleets supplied in the request.
+
+The request also lists fixed endpoint constraints such as max price, spot
+policy, backends, regions, instance types, fleets, env keys, tags, or backend
+options. Do not submit a task or service that conflicts with those values.
+
+Put each fixed constraint that has a dstack service YAML field into the final
+service YAML. For example, if the endpoint is limited to a fleet, max price, or
+spot policy, the final `service_yaml` should include the corresponding
+`service_yaml.fleets`, `service_yaml.max_price`, or `service_yaml.spot_policy`
+fields. Use `/dstack` for exact field names.
+
+If you cannot submit a useful task or service within the allowed fleets and
+constraints, write a failed `final_report.json`. The
+`final_report.json.failure_summary` value should say which fleet or constraint
+blocked the run and what the user/admin would need to change.
+
+# Task Usage
+
+Use `/dstack-prototyping` to learn how to use tasks. Keep in mind that using a
+task is a must. This means `sleep infinity` and directly attaching inside the
+task via SSH to run commands as the only way to use tasks. If there is any
+ambiguity, follow `/dstack-prototyping` and do nothing that contradicts it.
+
+# Backend/Fleet Selection, Idle Duration, Instance Volumes
+
+Use only the fleets allowed by the endpoint request. If the endpoint request
+also has constraints such as `backends`, `regions`, `instance_types`,
+`spot_policy`, or `max_price`, apply them when running `dstack offer` if the CLI
+has matching flags, and always apply them when submitting runs.
+
+## Backend And Fleet Choice
+
+Use `/dstack-prototyping` to learn how to select backends and fleets.
+
+Follow `/dstack-prototyping` skill on using a VM-based backend, Kubernetes backend, or SSH fleet that supports idle instances/instance volumes if there is such an option for the required GPU class.
+
+If backend allows (see above), use instance volumes to mount cache and model weights between runs.
+
+## Validating Offers
+
+When selecting backends/fleets and evaluating hardware that can be used to run
+the model, use `dstack offer --json` and pass `--fleet` explicitly. If the
+endpoint request has `fleets`, pass those fleets. Otherwise, pass every existing
+fleet. If `--fleet` is not passed, `dstack offer` can show offers that are not
+applicable to the fleets allowed by the endpoint request. Use these offers when
+selecting fleet, backend, and hardware.
+
+Choosing fleet/backend is gated by classifying each against `https://dstack.ai/docs/concepts/backends.md`. VM-based backends are listed under `## VM-based` (they support idle instances and instance volumes). Kubernetes backend is listed under `## Container-based`, but supports instance volumes and thus is preferred over other container-based backends.
+SSH fleets can be treated as VM-based backends as they support both idle instances (its equivalent) and instance volumes.
+
+## Model, Image, Serving Configuration, And Compute Fit
+
+Choose the model variant when allowed, image, serving configuration, and compute
+to optimize expected performance within endpoint constraints and current offers.
+
+If `Endpoint context:` contains `model_repo`, choose fleet/backend/hardware that
+can run `model_repo` within endpoint constraints.
+
+If `Endpoint context:` contains `base_model`, choose the repo/path and compute
+together. You may pick the base repo or a compatible variant that fits the
+allowed fleets, backends, and offers.
+
+If a task or service shows that the selected repo/path is a bad fit and
+`Endpoint context:` contains `base_model`, pick another compatible variant if
+available and test it in a task before submitting another service.
+
+## Submitting run
+
+When submitting a task or service, pass exact `fleets`, `backends`, and an
+intentional `resources` range based on the choice made from offers, so the run
+does not land outside the intended fleet/backend/hardware.
+
+## Decision Progress
+
+Progress should explain meaningful decisions and actions.
+
+When choosing a repo/path for `base_model`, fleet, backend, or hardware, write a
+progress message that includes:
+
+- `service_model_name` and the selected repo/path, when they differ;
+- the selected fleet, backend, and resource range;
+- the offer/docs evidence used for the choice;
+- the viable alternatives not selected and the exact reason;
+- the fleet, backend, and resources that will be used in the submitted YAML.
+- how the selected and rejected fleets/backends were classified;
+
+Backend-choice progress example (provide the same level of explanation for
+repo/path, fleet, and hardware choice):
+
+"I chose backend ... on fleet ... because ...; I did not choose ... because ...;
+I will submit YAML with fleets=..., backends=..., resources=...."
+
+# Final Service
+
+The final `service_yaml` is used to build the endpoint preset recipe. It must
+contain the full service config: `type: service`, the final run name, the service
+model name, image/commands/port, resources, env references, and the fixed
+endpoint constraints that apply to service YAML.
+
+Set final `service_yaml.name` to the final service run name.
+If final `service_yaml.model` is a string, set it to `service_model_name` from
+`Endpoint context:`. If final `service_yaml.model` is an object, set
+`service_yaml.model.name` to `service_model_name`.
+
+Before submitting the final service, choose service resources from the least
+restrictive requirements supported by the evidence, not from the exact machine
+that happened to run. For example, if the model worked on an A40 but the
+evidence only says it needs an NVIDIA GPU with at least 16GB memory, use that
+broader requirement. Use an exact GPU, region, backend, or instance type only
+when the endpoint constraints require it or the tested recipe depends on that
+exact choice.
+
+Use run status to know whether the final service is still starting, running, or
+failed. Use logs to understand failures. When dstack exposes the final service
+run's `service.url`, build its absolute URL using `DSTACK_ENDPOINT_SERVER_URL`
+and send a real model request using `DSTACK_ENDPOINT_BEARER_TOKEN` as the bearer
+token.
+
+Verify the context length that the final service actually supports and report
+it as `final_report.json.context_length`.
+
+# Benchmark
+
+Run one bounded benchmark using the same absolute dstack service URL and bearer
+authentication used for final service verification. Do not benchmark localhost.
+Run the benchmark client on the agent host, not inside the final service run.
+
+Use the serving framework's benchmark tool when applicable; otherwise use
+another trusted benchmark tool. Do not create or download a custom dataset.
+
+Write benchmark results to `benchmarks.jsonl`. Each line must include
+`success`, `run_name`, `run_type: service`, `tool`, `command`, `workload`, and
+either `metrics` or `failure_summary`.
+
+`tool` identifies the benchmark entrypoint without its version or arguments,
+for example `vllm bench serve`. Put its version in `tool_version` and the exact
+invocation in `command`.
+
+Configure and record an explicit maximum concurrency. `workload` must include
+`dataset_name`, `streaming`, `max_concurrency`, `request_rate`, `num_prompts`,
+`input_tokens`, and `output_tokens`. It may also include `ignore_eos` and
+`random_range_ratio`.
+
+Example format (values are illustrative):
+
+```json
+{"success":true,"run_name":"endpoint-1","run_type":"service","tool":"vllm bench serve","tool_version":"0.11.0","command":"vllm bench serve ...","workload":{"dataset_name":"random","streaming":true,"max_concurrency":1,"request_rate":"inf","num_prompts":16,"input_tokens":1024,"output_tokens":128},"metrics":{"completed":16,"output_throughput":42.1,"mean_ttft_ms":110.9,"p99_ttft_ms":121.6}}
+```
+
+Set `success` to `true` only when every benchmark request succeeds.
+
+Do not invent missing metrics or percentiles.
+
+Choose the workload before starting the benchmark and do not retry with a
+different workload. If benchmarking fails, write a failed benchmark record and
+a failed `final_report.json`.
+
+Do not write a successful `final_report.json` until the final service answers a
+request using `service_model_name` from `Endpoint context:` through the dstack
+service URL.
+
+# Secrets
+
+The dstack CLI is already configured. Do not inspect `~/.dstack/config.yml` or
+print, copy, or summarize tokens, secrets, or environment variable values.
+
+Do not expose the value of `DSTACK_TOKEN`, `DSTACK_ENDPOINT_BEARER_TOKEN`, or
+the value of any environment variable named by `Endpoint env keys available in
+the agent environment:` under `Fixed endpoint constraints:`.
+
+Do not put secret values in `final_report.json` or `benchmarks.jsonl`, and do
+not print them because stdout and stderr are stored in `agent_stdout.jsonl` and
+`agent_stderr.jsonl`. Use env references in `final_report.json.service_yaml`;
+use environment variable names or redacted values in
+`benchmarks.jsonl.command`.
+
+# Resume
+
+On startup or resume, inspect `final_report.json` and `submissions.jsonl`
+before submitting anything new. If `final_report.json` already reports success
+for the current endpoint configuration, do not submit another run. Otherwise use
+the next run number and record why the old run is not enough.
+
+# Final Report
+
+On success, write `final_report.json`, then return the structured final report.
+The successful report must include:
+
+- `final_report.json.base`: the base model repo for the deployed model
+- `final_report.json.model`: the exact repo/path loaded by the final service
+ command
+- `final_report.json.context_length`: the context length verified for the final
+ service
+
+Set `final_report.json.base` as follows:
+
+- If `Endpoint context:` contains `base_model`, set `final_report.json.base` to
+ `base_model`.
+- If `Endpoint context:` contains `model_repo`, inspect the repo metadata, model
+ card, config, or another reliable source to identify the base model repo.
+- If `model_repo` is itself the base model repo, set `final_report.json.base` to
+ `model_repo`.
+- Do not infer `final_report.json.base` only from the repo name.
+
+On failure, write `final_report.json` with a useful
+`final_report.json.failure_summary`, then return the structured final report.
+
+`final_report.json` must contain only the schema fields: `success`, `run_id`,
+`run_name`, `service_yaml`, `base`, `model`, `context_length`, and
+`failure_summary`.
+
+Stop after one correct service is verified and benchmarked. P/D disaggregation
+is not covered by the current endpoint agent or `/dstack-prototyping` skill.
diff --git a/src/dstack/_internal/core/services/endpoint_agent.py b/src/dstack/_internal/core/services/endpoint_agent.py
new file mode 100644
index 0000000000..a181e266cc
--- /dev/null
+++ b/src/dstack/_internal/core/services/endpoint_agent.py
@@ -0,0 +1,64 @@
+import enum
+import json
+from pathlib import Path
+from typing import Any, Sequence
+
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.profiles import ProfileParams
+
+_SYSTEM_PROMPT_PATH = (
+ Path(__file__).resolve().parent.parent / "resources" / "endpoint_agent" / "system_prompt.md"
+)
+
+
+def get_endpoint_agent_system_prompt() -> str:
+ return _SYSTEM_PROMPT_PATH.read_text(encoding="utf-8").strip()
+
+
+def format_endpoint_constraints(
+ configuration: EndpointConfiguration,
+ endpoint_env: dict[str, str],
+ *,
+ allowed_fleets: Sequence[str],
+) -> str:
+ lines = [
+ "Fixed endpoint constraints:",
+ "- Do not submit any task or service that conflicts with these values.",
+ "- Put applicable fixed constraints into the final service YAML.",
+ ]
+ for field in ProfileParams.__fields__:
+ if field == "fleets":
+ continue
+ value = getattr(configuration, field)
+ if value is not None:
+ lines.append(f"- {field}: {_format_constraint_value(value)}")
+ if allowed_fleets:
+ lines.append(f"- fleets: {', '.join(allowed_fleets)}")
+ if configuration.gateway is not None:
+ lines.append(f"- gateway: {_format_constraint_value(configuration.gateway)}")
+ lines.append(
+ "- Endpoint env keys available in the agent environment: "
+ + (", ".join(endpoint_env) if endpoint_env else "none")
+ )
+ return "\n".join(lines)
+
+
+def _format_constraint_value(value: Any) -> str:
+ data = _constraint_value_to_data(value)
+ if isinstance(data, (bool, dict, list)):
+ return json.dumps(data, sort_keys=True)
+ return str(data)
+
+
+def _constraint_value_to_data(value: Any) -> Any:
+ if isinstance(value, enum.Enum):
+ return value.value
+ if hasattr(value, "format"):
+ return value.format()
+ if hasattr(value, "json"):
+ return json.loads(value.json(exclude_none=True))
+ if isinstance(value, dict):
+ return {key: _constraint_value_to_data(item) for key, item in value.items()}
+ if isinstance(value, (list, tuple)):
+ return [_constraint_value_to_data(item) for item in value]
+ return value
diff --git a/src/dstack/_internal/core/services/endpoint_presets.py b/src/dstack/_internal/core/services/endpoint_presets.py
new file mode 100644
index 0000000000..b6da7d4c97
--- /dev/null
+++ b/src/dstack/_internal/core/services/endpoint_presets.py
@@ -0,0 +1,224 @@
+import hashlib
+import json
+import re
+from typing import Any
+
+import gpuhunt
+
+from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.endpoint_presets import (
+ EndpointBenchmark,
+ EndpointPresetRecipe,
+ EndpointPresetValidation,
+ EndpointPresetValidationReplica,
+)
+from dstack._internal.core.models.envs import EnvSentinel
+from dstack._internal.core.models.instances import Resources
+from dstack._internal.core.models.profiles import ProfileParams
+from dstack._internal.core.models.resources import ResourcesSpec
+from dstack._internal.utils.common import format_mib_as_gb
+
+_SECRET_ENV_PATTERN = re.compile(r"(token|key|secret|password)", re.IGNORECASE)
+
+
+def build_endpoint_preset_recipe(
+ *,
+ service: ServiceConfiguration,
+ validation_replicas: list[EndpointPresetValidationReplica],
+ base_model: str,
+ recipe_model: str,
+ context_length: int,
+ benchmarks: list[EndpointBenchmark],
+) -> EndpointPresetRecipe:
+ service = service.copy(deep=True)
+ service.name = None
+ service.gateway = None
+ for field in ProfileParams.__fields__:
+ setattr(service, field, None)
+ validation = EndpointPresetValidation(
+ replicas=validation_replicas,
+ benchmarks=benchmarks,
+ )
+ set_service_gpu_vendors_from_validations(service, [validation])
+ return EndpointPresetRecipe(
+ base=base_model,
+ id=make_endpoint_preset_recipe_id(service, context_length=context_length),
+ model=recipe_model,
+ context_length=context_length,
+ service=service,
+ validations=[validation],
+ )
+
+
+def make_endpoint_preset_recipe_id(
+ service: ServiceConfiguration,
+ context_length: int,
+) -> str:
+ payload = json.dumps(
+ {
+ "service": service_configuration_to_preset_data(service),
+ "context_length": context_length,
+ },
+ sort_keys=True,
+ separators=(",", ":"),
+ )
+ return hashlib.sha256(payload.encode()).hexdigest()[:8]
+
+
+def endpoint_preset_recipe_to_data(recipe: EndpointPresetRecipe) -> dict[str, Any]:
+ return {
+ "base": recipe.base,
+ "id": recipe.id,
+ "model": recipe.model,
+ "context_length": recipe.context_length,
+ "service": service_configuration_to_preset_data(recipe.service),
+ "validations": [
+ json.loads(validation.json(exclude_none=True)) for validation in recipe.validations
+ ],
+ }
+
+
+def service_configuration_to_preset_data(
+ configuration: ServiceConfiguration,
+) -> dict[str, Any]:
+ service_data = json.loads(configuration.json(exclude_none=True))
+ service_data.pop("type", None)
+ service_data.pop("name", None)
+ service_data.pop("gateway", None)
+ for field in ProfileParams.__fields__:
+ service_data.pop(field, None)
+ if configuration.env:
+ service_data["env"] = [
+ _env_item_to_preset_data(key, value)
+ for key, value in sorted(configuration.env.items())
+ ]
+ else:
+ service_data.pop("env", None)
+ for field, value in list(service_data.items()):
+ if value in ({}, []):
+ service_data.pop(field)
+ return service_data
+
+
+def resources_spec_from_instance_resources(resources: Resources) -> ResourcesSpec:
+ data: dict[str, Any] = {
+ "cpu": str(resources.cpus),
+ "memory": format_mib_as_gb(resources.memory_mib),
+ "disk": format_mib_as_gb(resources.disk.size_mib),
+ }
+ if resources.cpu_arch is not None:
+ data["cpu"] = f"{resources.cpu_arch.value}:{resources.cpus}"
+ if resources.gpus:
+ first_gpu = resources.gpus[0]
+ if any(
+ gpu.name != first_gpu.name
+ or gpu.memory_mib != first_gpu.memory_mib
+ or gpu.vendor != first_gpu.vendor
+ for gpu in resources.gpus
+ ):
+ raise ValueError("endpoint preset cannot be built from mixed-GPU instances")
+ data["gpu"] = {
+ "name": first_gpu.name,
+ "memory": format_mib_as_gb(first_gpu.memory_mib),
+ "count": len(resources.gpus),
+ }
+ if first_gpu.vendor is not None:
+ data["gpu"]["vendor"] = first_gpu.vendor.value
+ else:
+ data["gpu"] = 0
+ return ResourcesSpec.parse_obj(data)
+
+
+def set_service_gpu_vendors_from_validations(
+ service: ServiceConfiguration,
+ validations: list[EndpointPresetValidation],
+) -> None:
+ for group_num, group in enumerate(service.replica_groups):
+ resources = group.resources
+ if resources is None or not _requires_gpu(resources):
+ continue
+ validation_vendor = _get_validation_group_gpu_vendor(validations, group_num)
+ if validation_vendor is None or resources.gpu is None:
+ continue
+ if resources.gpu.vendor is not None and resources.gpu.vendor != validation_vendor:
+ raise ValueError("preset service GPU vendor does not match validation")
+ group_resources = _get_service_group_resources(service, group_num)
+ if group_resources.gpu is not None:
+ group_resources.gpu.vendor = validation_vendor
+
+
+def _env_item_to_preset_data(key: str, value: str | EnvSentinel) -> str:
+ if isinstance(value, EnvSentinel) or _is_secret_like_env(key, value):
+ return key
+ return f"{key}={value}"
+
+
+def _is_secret_like_env(key: str, value: str | EnvSentinel) -> bool:
+ if _SECRET_ENV_PATTERN.search(key):
+ return True
+ return isinstance(value, str) and _SECRET_ENV_PATTERN.search(value) is not None
+
+
+def _get_validation_group_gpu_vendor(
+ validations: list[EndpointPresetValidation],
+ group_num: int,
+) -> gpuhunt.AcceleratorVendor | None:
+ vendors = {
+ vendor
+ for validation in validations
+ for resources in validation.replicas[group_num].resources
+ if (vendor := _get_resources_gpu_vendor(resources)) is not None
+ }
+ if len(vendors) > 1:
+ raise ValueError("preset validations must not mix GPU vendors in a replica group")
+ return next(iter(vendors), None)
+
+
+def _get_resources_gpu_vendor(resources: ResourcesSpec) -> gpuhunt.AcceleratorVendor | None:
+ gpu = resources.gpu
+ if gpu is None or gpu.count.min == 0:
+ return None
+ if gpu.vendor is not None:
+ return gpu.vendor
+ if not gpu.name:
+ return None
+ vendors = {_get_gpu_name_vendor(name) for name in gpu.name} - {None}
+ if len(vendors) > 1:
+ raise ValueError("preset validations must not mix GPU vendors in a replica group")
+ return next(iter(vendors), None)
+
+
+def _get_gpu_name_vendor(name: str) -> gpuhunt.AcceleratorVendor | None:
+ known = (
+ (gpuhunt.KNOWN_NVIDIA_GPUS, gpuhunt.AcceleratorVendor.NVIDIA),
+ (gpuhunt.KNOWN_AMD_GPUS, gpuhunt.AcceleratorVendor.AMD),
+ (gpuhunt.KNOWN_INTEL_ACCELERATORS, gpuhunt.AcceleratorVendor.INTEL),
+ (gpuhunt.KNOWN_TENSTORRENT_ACCELERATORS, gpuhunt.AcceleratorVendor.TENSTORRENT),
+ )
+ for accelerators, vendor in known:
+ if any(accelerator.name.lower() == name.lower() for accelerator in accelerators):
+ return vendor
+ if name.startswith("tpu-"):
+ return gpuhunt.AcceleratorVendor.GOOGLE
+ return None
+
+
+def _get_service_group_resources(
+ service: ServiceConfiguration,
+ group_num: int,
+) -> ResourcesSpec:
+ resources = (
+ service.replicas[group_num].resources
+ if isinstance(service.replicas, list)
+ else service.resources
+ )
+ if resources is None:
+ raise ValueError("preset service object must specify resources")
+ return resources
+
+
+def _requires_gpu(resources: ResourcesSpec) -> bool:
+ gpu = resources.gpu
+ if gpu is None or gpu.count.max == 0:
+ return False
+ return gpu.count.min != 0 or gpu.count.max is not None
diff --git a/src/dstack/_internal/utils/ssh.py b/src/dstack/_internal/utils/ssh.py
index e1828bc33b..8b180127e1 100644
--- a/src/dstack/_internal/utils/ssh.py
+++ b/src/dstack/_internal/utils/ssh.py
@@ -101,12 +101,12 @@ def normalize_path(path: PathLike, *, collapse_user: bool = False) -> str:
:param collapse_user: try to replace user home prefix with `~`. `False` by default.
:return: Normalized path as string
"""
- if collapse_user:
+ if collapse_user and (openssh_home := _get_openssh_home()) is not None:
# The following "reverse" expanduser operation not only makes paths shorter and "nicer",
# but also fixes one specific issue with OpenSSH bundled with Git for Windows (MSYS2),
# see :func:`include_ssh_config` for details.
try:
- path = Path(path).relative_to(Path.home())
+ path = Path(path).relative_to(openssh_home)
path = f"~/{path}"
except ValueError:
pass
@@ -130,6 +130,20 @@ def normalize_path(path: PathLike, *, collapse_user: bool = False) -> str:
return str(path)
+def _get_openssh_home() -> Optional[Path]:
+ if IS_WINDOWS:
+ return Path.home()
+
+ # POSIX OpenSSH expands `~` from the passwd entry, even when `HOME` is overridden.
+ # Match that behavior so paths under a temporary `HOME` remain absolute in SSH config.
+ import pwd
+
+ try:
+ return Path(pwd.getpwuid(os.getuid()).pw_dir)
+ except KeyError:
+ return None
+
+
def include_ssh_config(path: PathLike, ssh_config_path: PathLike = default_ssh_config_path):
"""
Adds Include entry on top of the default ssh config file
diff --git a/src/tests/_internal/cli/commands/test_endpoint.py b/src/tests/_internal/cli/commands/test_endpoint.py
new file mode 100644
index 0000000000..14f908f6ab
--- /dev/null
+++ b/src/tests/_internal/cli/commands/test_endpoint.py
@@ -0,0 +1,83 @@
+from unittest.mock import patch
+
+import pytest
+
+from dstack._internal.cli.services.endpoint_presets import EndpointPresetStore
+from tests._internal.cli.common import run_dstack_cli
+from tests._internal.cli.endpoint_presets import get_endpoint_preset_recipe
+
+pytestmark = pytest.mark.windows
+
+
+class TestEndpointPresetLocalCommands:
+ def test_handles_keyboard_interrupt(self, tmp_path, capsys):
+ configuration_path = tmp_path / "endpoint.dstack.yml"
+ configuration_path.write_text(
+ "type: endpoint\nname: qwen\nmodel:\n base: Qwen/Qwen3.5-27B\n"
+ )
+
+ with (
+ patch("dstack.api.Client.from_config"),
+ patch(
+ "dstack._internal.cli.commands.endpoint.create_endpoint_preset",
+ side_effect=KeyboardInterrupt,
+ ),
+ ):
+ exit_code = run_dstack_cli(
+ ["endpoint", "preset", "create", "-f", str(configuration_path)],
+ home_dir=tmp_path,
+ )
+
+ assert exit_code == 0
+ assert "Operation interrupted by user" in capsys.readouterr().out
+
+ def test_lists_and_deletes_recipe_without_api_client(self, tmp_path, capsys):
+ recipe = get_endpoint_preset_recipe()
+ EndpointPresetStore(tmp_path / ".dstack" / "presets").save(recipe)
+
+ with patch("dstack.api.Client.from_config") as from_config:
+ assert run_dstack_cli(["endpoint", "preset", "list"], home_dir=tmp_path) == 0
+ from_config.assert_not_called()
+
+ output = capsys.readouterr().out
+ assert "Qwen/Qwen3.5-27B" in output
+ assert "recipe=8f3a12c4" in output
+ assert "repo=community/Qwen3.5-27B-GPTQ-Int4" in output
+
+ with patch("dstack.api.Client.from_config") as from_config:
+ assert (
+ run_dstack_cli(
+ [
+ "endpoint",
+ "preset",
+ "delete",
+ "--recipe",
+ recipe.id,
+ "-y",
+ ],
+ home_dir=tmp_path,
+ )
+ == 0
+ )
+ from_config.assert_not_called()
+
+ assert EndpointPresetStore(tmp_path / ".dstack" / "presets").list() == []
+ assert not (tmp_path / ".dstack" / "presets" / "models--Qwen--Qwen3.5-27B").exists()
+
+ def test_deletes_preset_without_api_client(self, tmp_path):
+ recipe = get_endpoint_preset_recipe()
+ store = EndpointPresetStore(tmp_path / ".dstack" / "presets")
+ store.save(recipe)
+ store.save(recipe.copy(update={"id": "01234567"}))
+
+ with patch("dstack.api.Client.from_config") as from_config:
+ assert (
+ run_dstack_cli(
+ ["endpoint", "preset", "delete", recipe.base, "-y"],
+ home_dir=tmp_path,
+ )
+ == 0
+ )
+ from_config.assert_not_called()
+
+ assert store.list() == []
diff --git a/src/tests/_internal/cli/common.py b/src/tests/_internal/cli/common.py
index 09f4541c7e..b5f1dd915c 100644
--- a/src/tests/_internal/cli/common.py
+++ b/src/tests/_internal/cli/common.py
@@ -4,6 +4,7 @@
from unittest.mock import patch
from dstack._internal.cli.main import main
+from dstack._internal.compat import IS_WINDOWS
def run_dstack_cli(
@@ -16,8 +17,11 @@ def run_dstack_cli(
cwd = os.getcwd()
os.chdir(repo_dir)
if home_dir is not None:
- prev_home_dir = os.environ["HOME"]
+ prev_home_dir = os.environ.get("HOME")
os.environ["HOME"] = str(home_dir)
+ if IS_WINDOWS:
+ prev_userprofile = os.environ.get("USERPROFILE")
+ os.environ["USERPROFILE"] = str(home_dir)
with patch("sys.argv", ["dstack"] + cli_args):
try:
main()
@@ -25,7 +29,15 @@ def run_dstack_cli(
exit_code = e.code
finally:
if home_dir is not None:
- os.environ["HOME"] = prev_home_dir
+ if prev_home_dir is None:
+ os.environ.pop("HOME", None)
+ else:
+ os.environ["HOME"] = prev_home_dir
+ if IS_WINDOWS:
+ if prev_userprofile is None:
+ os.environ.pop("USERPROFILE", None)
+ else:
+ os.environ["USERPROFILE"] = prev_userprofile
if repo_dir is not None:
os.chdir(cwd)
return exit_code
diff --git a/src/tests/_internal/cli/endpoint_presets.py b/src/tests/_internal/cli/endpoint_presets.py
new file mode 100644
index 0000000000..8ae4301a00
--- /dev/null
+++ b/src/tests/_internal/cli/endpoint_presets.py
@@ -0,0 +1,137 @@
+from types import SimpleNamespace
+from uuid import uuid4
+
+from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.endpoint_agent import AgentFinalReport
+from dstack._internal.core.models.endpoint_presets import (
+ EndpointBenchmark,
+ EndpointBenchmarkClient,
+ EndpointBenchmarkTarget,
+ EndpointPresetRecipe,
+ EndpointPresetValidation,
+ EndpointPresetValidationReplica,
+)
+from dstack._internal.core.models.instances import Disk, Gpu, Resources
+from dstack._internal.core.models.resources import ResourcesSpec
+from dstack._internal.core.models.runs import JobStatus, Run, RunStatus, ServiceSpec
+
+
+def get_endpoint_benchmark(*, run_id=None, run_name: str = "qwen-build-2") -> EndpointBenchmark:
+ return EndpointBenchmark(
+ success=True,
+ run_id=run_id,
+ run_name=run_name,
+ run_type="service",
+ tool="vllm bench serve",
+ command="vllm bench serve --base-url $SERVICE_URL",
+ workload={
+ "dataset_name": "random",
+ "streaming": True,
+ "max_concurrency": 1,
+ "request_rate": "inf",
+ "num_prompts": 16,
+ "input_tokens": 1024,
+ "output_tokens": 128,
+ },
+ metrics={"output_throughput": 42.1},
+ target=EndpointBenchmarkTarget(type="server-proxy"),
+ client=EndpointBenchmarkClient(type="local"),
+ )
+
+
+def get_endpoint_preset_recipe(
+ *,
+ recipe_id: str = "8f3a12c4",
+ context_length: int = 32768,
+) -> EndpointPresetRecipe:
+ resources = ResourcesSpec.parse_obj(
+ {
+ "cpu": "16",
+ "memory": "64GB",
+ "disk": "200GB",
+ "gpu": {"name": "A6000", "memory": "48GB", "count": 1},
+ }
+ )
+ return EndpointPresetRecipe(
+ base="Qwen/Qwen3.5-27B",
+ id=recipe_id,
+ model="community/Qwen3.5-27B-GPTQ-Int4",
+ context_length=context_length,
+ service=ServiceConfiguration.parse_obj(
+ {
+ "image": "vllm/vllm-openai:v0.11.0",
+ "commands": ["vllm serve community/Qwen3.5-27B-GPTQ-Int4"],
+ "port": 8000,
+ "model": "Qwen/Qwen3.5-27B",
+ "resources": {"gpu": "nvidia:40GB..48GB:1"},
+ "env": ["HF_TOKEN"],
+ }
+ ),
+ validations=[
+ EndpointPresetValidation(
+ replicas=[EndpointPresetValidationReplica(resources=[resources])],
+ benchmarks=[get_endpoint_benchmark()],
+ )
+ ],
+ )
+
+
+def get_running_service_run() -> Run:
+ service = ServiceConfiguration.parse_obj(
+ {
+ "name": "qwen-build-2",
+ "image": "vllm/vllm-openai:v0.11.0",
+ "commands": [
+ "vllm serve community/Qwen3.5-27B-GPTQ-Int4 --served-model-name Qwen/Qwen3.5-27B"
+ ],
+ "port": 8000,
+ "model": "Qwen/Qwen3.5-27B",
+ "gateway": "benchmark-gateway",
+ "fleets": ["gpu-fleet"],
+ "backends": ["verda"],
+ "spot_policy": "auto",
+ "max_price": 0.5,
+ "env": {"LICENSE": "license-secret"},
+ "resources": {"gpu": "40GB..48GB:1"},
+ }
+ )
+ resources = Resources(
+ cpus=16,
+ memory_mib=64 * 1024,
+ gpus=[Gpu(name="A6000", memory_mib=48 * 1024)],
+ spot=False,
+ disk=Disk(size_mib=200 * 1024),
+ )
+ job = SimpleNamespace(
+ job_spec=SimpleNamespace(job_num=0, replica_num=0, replica_group="0"),
+ job_submissions=[
+ SimpleNamespace(
+ deployment_num=0,
+ status=JobStatus.RUNNING,
+ job_runtime_data=SimpleNamespace(
+ offer=SimpleNamespace(instance=SimpleNamespace(resources=resources))
+ ),
+ )
+ ],
+ )
+ return Run.construct(
+ id=uuid4(),
+ project_name="main",
+ status=RunStatus.RUNNING,
+ run_spec=SimpleNamespace(run_name="qwen-build-2", configuration=service),
+ jobs=[job],
+ service=ServiceSpec(url="/proxy/services/main/qwen-build-2/"),
+ deployment_num=0,
+ )
+
+
+def get_successful_endpoint_report(run: Run) -> AgentFinalReport:
+ return AgentFinalReport(
+ success=True,
+ run_id=run.id,
+ run_name=run.run_spec.run_name,
+ service_yaml="type: service",
+ base="Qwen/Qwen3.5-27B",
+ model="community/Qwen3.5-27B-GPTQ-Int4",
+ context_length=32768,
+ )
diff --git a/src/tests/_internal/cli/services/test_endpoint_agent_runtime.py b/src/tests/_internal/cli/services/test_endpoint_agent_runtime.py
new file mode 100644
index 0000000000..9cc0f09387
--- /dev/null
+++ b/src/tests/_internal/cli/services/test_endpoint_agent_runtime.py
@@ -0,0 +1,206 @@
+import asyncio
+import os
+import shutil
+import subprocess
+import sys
+from types import SimpleNamespace
+
+import psutil
+import pytest
+
+from dstack._internal.cli.services.endpoint_agent_runtime import (
+ ClaudeAuth,
+ EndpointAgentWorkspace,
+ _build_claude_command,
+ _prepare_subprocess_command,
+ _ProgressTailer,
+ _terminate_process,
+ build_endpoint_agent_env,
+ contains_redacted_value,
+ endpoint_agent_workspace,
+ get_claude_auth,
+ run_endpoint_agent,
+)
+from dstack._internal.compat import IS_WINDOWS
+from dstack._internal.core.errors import CLIError
+
+pytestmark = pytest.mark.windows
+
+
+def _claude_auth(*, use_existing: bool = False, effort=None) -> ClaudeAuth:
+ return ClaudeAuth(
+ api_key=None if use_existing else "anthropic-secret",
+ executable="claude",
+ effort=effort,
+ model="claude-test",
+ use_existing=use_existing,
+ )
+
+
+class TestClaudeAuth:
+ def test_rejects_api_key_and_existing_auth_together(self, monkeypatch):
+ monkeypatch.setenv("DSTACK_AGENT_ANTHROPIC_API_KEY", "key")
+ monkeypatch.setenv("DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH", "1")
+
+ with pytest.raises(CLIError, match="cannot both be set"):
+ get_claude_auth()
+
+ def test_requires_an_auth_mode(self, monkeypatch):
+ monkeypatch.delenv("DSTACK_AGENT_ANTHROPIC_API_KEY", raising=False)
+ monkeypatch.delenv("DSTACK_AGENT_CLAUDE_USE_EXISTING_AUTH", raising=False)
+
+ with pytest.raises(CLIError, match="DSTACK_AGENT_ANTHROPIC_API_KEY is not set"):
+ get_claude_auth()
+
+ @pytest.mark.parametrize("use_existing", [False, True])
+ def test_builds_command_for_selected_auth_mode(self, use_existing):
+ command = _build_claude_command(
+ auth=_claude_auth(use_existing=use_existing, effort="high")
+ )
+
+ assert ("--bare" in command) is not use_existing
+ assert ("--setting-sources" in command) is use_existing
+ assert command[command.index("--effort") + 1] == "high"
+
+ @pytest.mark.windows_only
+ def test_runs_windows_batch_launcher(self, tmp_path):
+ script = tmp_path / "fake-claude.cmd"
+ script.write_text("@echo off\necho batch-ok\n")
+
+ result = subprocess.run(
+ _prepare_subprocess_command([str(script)]),
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+
+ assert result.returncode == 0
+ assert result.stdout.strip() == "batch-ok"
+
+
+class TestAgentIsolation:
+ def test_inherits_only_required_environment(self, tmp_path, monkeypatch):
+ monkeypatch.setenv("PATH", "/usr/bin")
+ monkeypatch.setenv("HOME", "/home/test")
+ monkeypatch.setenv("UNRELATED_SECRET", "must-not-be-inherited")
+ api = SimpleNamespace(
+ project="main",
+ client=SimpleNamespace(base_url="http://127.0.0.1:3000"),
+ )
+
+ env = build_endpoint_agent_env(
+ api=api,
+ endpoint_env={"HF_TOKEN": "hf-secret"},
+ auth=_claude_auth(),
+ workspace=EndpointAgentWorkspace(
+ path=tmp_path,
+ dstack_home=tmp_path / "home",
+ ),
+ token="dstack-secret",
+ )
+
+ assert env["DSTACK_SERVER_URL"] == "http://127.0.0.1:3000"
+ assert env["DSTACK_PROJECT"] == "main"
+ assert env["DSTACK_TOKEN"] == "dstack-secret"
+ assert env["HF_TOKEN"] == "hf-secret"
+ assert "UNRELATED_SECRET" not in env
+
+ def test_creates_private_cli_home_and_dstack_wrapper(self):
+ with endpoint_agent_workspace() as workspace:
+ if not IS_WINDOWS:
+ assert (workspace.dstack_home / ".ssh").stat().st_mode & 0o777 == 0o700
+ assert not (workspace.dstack_home / ".dstack" / "config.yml").exists()
+ dstack_command = shutil.which("dstack", path=str(workspace.bin_path))
+ assert dstack_command is not None
+ result = subprocess.run(
+ [dstack_command, "--help"],
+ check=False,
+ capture_output=True,
+ text=True,
+ )
+ assert result.returncode == 0
+ assert "Usage: dstack" in result.stdout
+
+ def test_keeps_control_socket_path_bounded(self):
+ with endpoint_agent_workspace() as workspace:
+ if not IS_WINDOWS:
+ socket_path = (
+ workspace.dstack_home / ".dstack" / "ssh" / f"{'x' * 41}.control.sock"
+ )
+ assert len(os.fsencode(socket_path)) <= 103
+
+ def test_detects_known_secret_in_generated_artifact(self):
+ assert contains_redacted_value(
+ {"commands": ["serve --token secret-token"]},
+ ("secret-token",),
+ )
+
+
+class TestAgentOutput:
+ @pytest.mark.asyncio
+ async def test_sends_prompt_and_redacts_raw_output(self, tmp_path, monkeypatch):
+ script = tmp_path / "fake_claude.py"
+ script.write_text(
+ """import json
+import sys
+
+prompt = sys.stdin.read()
+print(json.dumps({
+ "type": "result",
+ "is_error": True,
+ "result": "bad secret-token",
+ "structured_output": {"prompt": prompt},
+}))
+"""
+ )
+ (tmp_path / "progress.jsonl").touch()
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_agent_runtime._build_claude_command",
+ lambda **_: [sys.executable, str(script)],
+ )
+
+ workspace = EndpointAgentWorkspace(path=tmp_path, dstack_home=tmp_path / "home")
+ output = await run_endpoint_agent(
+ prompt="full endpoint prompt",
+ env=os.environ.copy(),
+ workspace=workspace,
+ auth=_claude_auth(),
+ redacted_values=("secret-token",),
+ )
+
+ assert output.report_data == {"prompt": "full endpoint prompt"}
+ assert output.error == "bad [redacted]"
+ assert "secret-token" not in workspace.stdout_path.read_text()
+
+ def test_progress_stream_prints_only_redacted_messages(self, tmp_path, capsys):
+ progress_path = tmp_path / "progress.jsonl"
+ progress_path.write_text('{"message":"using secret-token"}\n')
+
+ _ProgressTailer(path=progress_path, redacted_values=("secret-token",)).flush()
+
+ output = capsys.readouterr().out
+ assert "using [redacted]" in output
+ assert "secret-token" not in output
+
+
+class TestProcessCleanup:
+ @pytest.mark.windows_only
+ @pytest.mark.asyncio
+ async def test_terminates_windows_process_tree(self):
+ proc = await asyncio.create_subprocess_exec(
+ sys.executable,
+ "-c",
+ (
+ "import subprocess,sys,time; "
+ "p=subprocess.Popen([sys.executable,'-c','import time; time.sleep(60)']); "
+ "print(p.pid,flush=True); time.sleep(60)"
+ ),
+ stdout=asyncio.subprocess.PIPE,
+ )
+ assert proc.stdout is not None
+ child_pid = int((await proc.stdout.readline()).decode())
+
+ await _terminate_process(proc)
+
+ assert proc.returncode is not None
+ assert not psutil.pid_exists(child_pid)
diff --git a/src/tests/_internal/cli/services/test_endpoint_preset_apply.py b/src/tests/_internal/cli/services/test_endpoint_preset_apply.py
new file mode 100644
index 0000000000..dbdd9f66fd
--- /dev/null
+++ b/src/tests/_internal/cli/services/test_endpoint_preset_apply.py
@@ -0,0 +1,139 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from dstack._internal.cli.services.endpoint_preset_apply import (
+ _build_service,
+ _get_matching_recipes,
+ _select_plan,
+ apply_endpoint_preset,
+)
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.instances import InstanceAvailability
+from tests._internal.cli.endpoint_presets import get_endpoint_preset_recipe
+
+pytestmark = pytest.mark.windows
+
+
+class TestGetMatchingRecipes:
+ def test_matches_base_model_context_and_recipe(self):
+ recipes = [
+ get_endpoint_preset_recipe(recipe_id="small", context_length=4096),
+ get_endpoint_preset_recipe(recipe_id="large", context_length=32768),
+ ]
+ configuration = EndpointConfiguration(
+ name="qwen",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ context_length=8192,
+ )
+
+ assert _get_matching_recipes(recipes, configuration=configuration, recipe_id=None) == [
+ recipes[1]
+ ]
+ assert _get_matching_recipes(recipes, configuration=configuration, recipe_id="large") == [
+ recipes[1]
+ ]
+ assert not _get_matching_recipes(recipes, configuration=configuration, recipe_id="small")
+
+ def test_exact_request_matches_repo_and_client_facing_name(self):
+ matching = get_endpoint_preset_recipe(recipe_id="matching")
+ configuration = EndpointConfiguration(
+ name="qwen",
+ model={
+ "repo": "community/Qwen3.5-27B-GPTQ-Int4",
+ "name": "Qwen/Qwen3.5-27B",
+ },
+ )
+
+ assert _get_matching_recipes([matching], configuration=configuration, recipe_id=None) == [
+ matching
+ ]
+ assert not _get_matching_recipes(
+ [matching.copy(update={"model": "other/repo"})],
+ configuration=configuration,
+ recipe_id=None,
+ )
+
+
+class TestBuildService:
+ def test_applies_endpoint_name_env_gateway_and_constraints(self):
+ configuration = EndpointConfiguration(
+ name="qwen-production",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ gateway="inference",
+ env={"HF_TOKEN": "token"},
+ fleets=["gpu-fleet"],
+ max_price=1,
+ )
+
+ service = _build_service(configuration, get_endpoint_preset_recipe())
+
+ assert service.name == "qwen-production"
+ assert service.gateway == "inference"
+ assert service.env["HF_TOKEN"] == "token"
+ assert [fleet.format() for fleet in service.fleets] == ["gpu-fleet"]
+ assert service.max_price == 1
+
+
+class TestSelectPlan:
+ def test_selects_first_recipe_with_available_offer(self):
+ recipes = [
+ get_endpoint_preset_recipe(recipe_id="unavailable"),
+ get_endpoint_preset_recipe(recipe_id="available"),
+ ]
+ configurator = Mock()
+ prepared = [
+ Mock(run_plan=_plan(InstanceAvailability.NOT_AVAILABLE)),
+ Mock(run_plan=_plan(InstanceAvailability.AVAILABLE)),
+ ]
+ configurator.prepare_configuration.side_effect = prepared
+ service_args = SimpleNamespace(profile=None)
+
+ selected = _select_plan(
+ configuration=EndpointConfiguration(name="qwen", model={"base": "Qwen/Qwen3.5-27B"}),
+ configuration_path="endpoint.dstack.yml",
+ recipes=recipes,
+ configurator=configurator,
+ service_args=service_args,
+ )
+
+ assert selected.recipe.id == "available"
+ assert selected.prepared is prepared[1]
+ assert configurator.prepare_configuration.call_count == 2
+
+ def test_applies_the_selected_prepared_plan(self, monkeypatch):
+ recipe = get_endpoint_preset_recipe()
+ prepared = Mock(run_plan=_plan(InstanceAvailability.AVAILABLE))
+ configurator = Mock()
+ configurator.get_parser.return_value.parse_args.return_value = SimpleNamespace()
+ configurator.prepare_configuration.return_value = prepared
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_apply.ServiceConfigurator",
+ lambda api_client: configurator,
+ )
+ command_args = SimpleNamespace()
+
+ apply_endpoint_preset(
+ api=Mock(),
+ configuration=EndpointConfiguration(
+ name="qwen",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ ),
+ configuration_path="endpoint.dstack.yml",
+ recipe_id=None,
+ command_args=command_args,
+ store=Mock(list=Mock(return_value=[recipe])),
+ )
+
+ configurator.apply_prepared_configuration.assert_called_once_with(
+ prepared=prepared,
+ command_args=command_args,
+ configurator_args=configurator.get_parser.return_value.parse_args.return_value,
+ )
+
+
+def _plan(availability: InstanceAvailability):
+ return SimpleNamespace(
+ job_plans=[SimpleNamespace(offers=[SimpleNamespace(availability=availability)])]
+ )
diff --git a/src/tests/_internal/cli/services/test_endpoint_preset_create.py b/src/tests/_internal/cli/services/test_endpoint_preset_create.py
new file mode 100644
index 0000000000..2f26d1ac70
--- /dev/null
+++ b/src/tests/_internal/cli/services/test_endpoint_preset_create.py
@@ -0,0 +1,269 @@
+import asyncio
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from dstack._internal.cli.services.endpoint_agent_runtime import (
+ ClaudeAuth,
+ EndpointAgentProcessOutput,
+ EndpointAgentWorkspace,
+)
+from dstack._internal.cli.services.endpoint_preset_create import (
+ _build_prompt,
+ _cleanup_runs,
+ _create_endpoint_preset,
+ _get_build_name,
+)
+from dstack._internal.cli.services.endpoint_presets import EndpointPresetStore
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.runs import Run, RunStatus
+from tests._internal.cli.endpoint_presets import (
+ get_endpoint_benchmark,
+ get_running_service_run,
+ get_successful_endpoint_report,
+)
+
+pytestmark = pytest.mark.windows
+
+
+def _claude_auth() -> ClaudeAuth:
+ return ClaudeAuth(
+ api_key="anthropic-secret",
+ executable="claude",
+ effort=None,
+ model="claude-test",
+ use_existing=False,
+ )
+
+
+@pytest.fixture
+def creation_context(tmp_path, monkeypatch):
+ run = get_running_service_run()
+ run_apis = _FakeRunAPIs(run)
+ api = SimpleNamespace(
+ project="main",
+ runs=run_apis,
+ client=SimpleNamespace(
+ _token="dstack-secret",
+ base_url="http://127.0.0.1:3000",
+ runs=run_apis,
+ ),
+ )
+ configuration = EndpointConfiguration(
+ name="qwen-build",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ context_length=8192,
+ fleets=["gpu-fleet"],
+ env={"LICENSE": "license-secret"},
+ )
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create.get_claude_auth",
+ _claude_auth,
+ )
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create._get_build_name",
+ lambda _: "qwen-build",
+ )
+ return SimpleNamespace(
+ api=api,
+ configuration=configuration,
+ run=run,
+ run_apis=run_apis,
+ store=EndpointPresetStore(tmp_path / "presets"),
+ )
+
+
+class TestCreateEndpointPreset:
+ @pytest.mark.asyncio
+ async def test_checks_active_fleets_before_claude_auth(self, tmp_path, monkeypatch):
+ api = SimpleNamespace(
+ project="main",
+ client=SimpleNamespace(fleets=SimpleNamespace(list=lambda *args, **kwargs: [])),
+ )
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create.get_claude_auth",
+ lambda: pytest.fail("Claude auth must not be checked without an active fleet"),
+ )
+
+ with pytest.raises(CLIError, match="no active fleets"):
+ await _create_endpoint_preset(
+ api=api,
+ configuration=EndpointConfiguration(
+ name="qwen-build",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ ),
+ store=EndpointPresetStore(tmp_path / "presets"),
+ )
+
+ @pytest.mark.asyncio
+ async def test_cleans_up_runs_when_cancelled(self, creation_context, monkeypatch):
+ async def run_agent(**_):
+ raise asyncio.CancelledError
+
+ cleanup_calls = []
+
+ async def cleanup_runs(**kwargs):
+ cleanup_calls.append(kwargs)
+
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create.run_endpoint_agent",
+ run_agent,
+ )
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create._cleanup_runs",
+ cleanup_runs,
+ )
+
+ with pytest.raises(asyncio.CancelledError):
+ await _create_endpoint_preset(
+ api=creation_context.api,
+ configuration=creation_context.configuration,
+ store=creation_context.store,
+ )
+
+ assert len(cleanup_calls) == 1
+ assert cleanup_calls[0]["build_name"] == "qwen-build"
+
+ @pytest.mark.parametrize(
+ ("keep_service", "stopped_names"),
+ [(False, ["qwen-build-2"]), (True, [])],
+ )
+ @pytest.mark.asyncio
+ async def test_saves_recipe_and_cleans_up_runs(
+ self, creation_context, monkeypatch, keep_service, stopped_names
+ ):
+ async def run_agent(**kwargs):
+ workspace = kwargs["workspace"]
+ benchmark = get_endpoint_benchmark(
+ run_id=creation_context.run.id,
+ run_name=creation_context.run.run_spec.run_name,
+ )
+ workspace.benchmarks_path.write_text(benchmark.json() + "\n")
+ return EndpointAgentProcessOutput(
+ report_data=json.loads(get_successful_endpoint_report(creation_context.run).json())
+ )
+
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create.run_endpoint_agent",
+ run_agent,
+ )
+ result = await _create_endpoint_preset(
+ api=creation_context.api,
+ configuration=creation_context.configuration,
+ store=creation_context.store,
+ keep_service=keep_service,
+ )
+
+ assert result.recipe.base == "Qwen/Qwen3.5-27B"
+ assert result.path.is_file()
+ assert creation_context.store.list() == [result.recipe]
+ assert "license-secret" not in result.path.read_text()
+ assert creation_context.run_apis.stopped_names == stopped_names
+
+
+class TestBuildName:
+ def test_requires_name_and_keeps_generated_prefix_bounded(self, monkeypatch):
+ with pytest.raises(CLIError, match="Endpoint name is required"):
+ _get_build_name(None)
+ monkeypatch.setattr(
+ "dstack._internal.cli.services.endpoint_preset_create.secrets.token_hex",
+ lambda _: "a1b2c3",
+ )
+
+ build_name = _get_build_name("qwen-endpoint-with-a-name-that-is-forty-one")
+
+ assert len(f"{build_name}-99999") <= 41
+
+
+class TestBuildPrompt:
+ def test_distinguishes_base_from_exact_model(self):
+ base_prompt = _build_prompt(
+ configuration=EndpointConfiguration(
+ name="qwen",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ context_length=8192,
+ ),
+ build_name="qwen-build",
+ allowed_fleets=("gpu-fleet",),
+ )
+ exact_prompt = _build_prompt(
+ configuration=EndpointConfiguration(
+ name="qwen",
+ model={
+ "repo": "community/Qwen3.5-27B-GPTQ-Int4",
+ "name": "Qwen/Qwen3.5-27B",
+ },
+ ),
+ build_name="qwen-build",
+ allowed_fleets=("gpu-fleet",),
+ )
+
+ assert "- base_model: Qwen/Qwen3.5-27B" in base_prompt
+ assert "- model_repo:" not in base_prompt
+ assert "- context_length: 8192" in base_prompt
+ assert "- model_repo: community/Qwen3.5-27B-GPTQ-Int4" in exact_prompt
+ assert "- base_model:" not in exact_prompt
+ assert "- context_length:" not in exact_prompt
+
+
+class TestCleanupRuns:
+ @pytest.mark.asyncio
+ async def test_stops_only_recorded_build_runs(self, tmp_path, monkeypatch):
+ (tmp_path / "submissions.jsonl").write_text(
+ '{"name":"qwen-build-1"}\n{"name":"unrelated-run"}\n{"name":"qwen-build-2"}\n'
+ )
+ runs = _FakeRuns()
+ api = SimpleNamespace(
+ runs=runs,
+ project="main",
+ client=SimpleNamespace(runs=runs),
+ )
+
+ async def no_sleep(_):
+ return None
+
+ monkeypatch.setattr(asyncio, "sleep", no_sleep)
+ await _cleanup_runs(
+ api=api,
+ build_name="qwen-build",
+ workspace=EndpointAgentWorkspace(
+ path=tmp_path,
+ dstack_home=tmp_path / "home",
+ ),
+ final_run_name="qwen-build-2",
+ keep_final_service=True,
+ )
+
+ assert runs.stopped_names == ["qwen-build-1"]
+
+
+class _FakeRuns:
+ def __init__(self):
+ self.stopped_names: list[str] = []
+
+ def get(self, name):
+ status = RunStatus.TERMINATED if name in self.stopped_names else RunStatus.RUNNING
+ return SimpleNamespace(status=status)
+
+ def stop(self, project, names, abort):
+ assert project == "main"
+ assert abort is False
+ self.stopped_names.extend(names)
+
+
+class _FakeRunAPIs:
+ def __init__(self, run: Run):
+ self.run = run
+ self.stopped_names: list[str] = []
+
+ def get(self, *args):
+ name = args[-1]
+ return self.run if name == self.run.run_spec.run_name else None
+
+ def stop(self, project, names, abort):
+ assert project == "main"
+ assert abort is False
+ self.stopped_names.extend(names)
+ self.run.status = RunStatus.TERMINATED
diff --git a/src/tests/_internal/cli/services/test_endpoint_preset_verify.py b/src/tests/_internal/cli/services/test_endpoint_preset_verify.py
new file mode 100644
index 0000000000..8bf51578c6
--- /dev/null
+++ b/src/tests/_internal/cli/services/test_endpoint_preset_verify.py
@@ -0,0 +1,65 @@
+import pytest
+
+from dstack._internal.cli.services.endpoint_preset_verify import (
+ build_verified_endpoint_preset,
+)
+from dstack._internal.core.errors import CLIError
+from dstack._internal.core.models.endpoints import EndpointConfiguration
+from dstack._internal.core.models.envs import EnvSentinel
+from dstack._internal.core.models.profiles import ProfileParams
+from tests._internal.cli.endpoint_presets import (
+ get_endpoint_benchmark,
+ get_running_service_run,
+ get_successful_endpoint_report,
+)
+
+pytestmark = pytest.mark.windows
+
+
+class TestBuildVerifiedEndpointPreset:
+ def test_builds_portable_self_contained_recipe(self):
+ run = get_running_service_run()
+
+ recipe = build_verified_endpoint_preset(
+ run=run,
+ endpoint_configuration=EndpointConfiguration(
+ name="qwen-build",
+ model={"base": "Qwen/Qwen3.5-27B"},
+ context_length=8192,
+ gateway="benchmark-gateway",
+ env={"LICENSE": "license-secret"},
+ ),
+ report=get_successful_endpoint_report(run),
+ benchmarks=[get_endpoint_benchmark(run_id=run.id, run_name=run.run_spec.run_name)],
+ )
+
+ assert recipe.base == "Qwen/Qwen3.5-27B"
+ assert recipe.model == "community/Qwen3.5-27B-GPTQ-Int4"
+ assert recipe.context_length == 32768
+ assert recipe.service.name is None
+ assert recipe.service.gateway is None
+ assert all(getattr(recipe.service, field) is None for field in ProfileParams.__fields__)
+ assert isinstance(recipe.service.env["LICENSE"], EnvSentinel)
+ assert recipe.service.resources.gpu.vendor.value == "nvidia"
+ validation = recipe.validations[0]
+ assert validation.replicas[0].resources[0].gpu.name == ["A6000"]
+ assert validation.benchmarks[0].target.type == "server-proxy"
+ assert validation.benchmarks[0].client.type == "local"
+
+ def test_rejects_variant_for_exact_model_request(self):
+ run = get_running_service_run()
+ report = get_successful_endpoint_report(run).copy(update={"model": "other/model"})
+
+ with pytest.raises(CLIError, match="changed an exact model request"):
+ build_verified_endpoint_preset(
+ run=run,
+ endpoint_configuration=EndpointConfiguration(
+ name="qwen-build",
+ model={
+ "repo": "community/Qwen3.5-27B-GPTQ-Int4",
+ "name": "Qwen/Qwen3.5-27B",
+ },
+ ),
+ report=report,
+ benchmarks=[get_endpoint_benchmark(run_id=run.id, run_name=run.run_spec.run_name)],
+ )
diff --git a/src/tests/_internal/cli/services/test_endpoint_presets.py b/src/tests/_internal/cli/services/test_endpoint_presets.py
new file mode 100644
index 0000000000..ef8aa584de
--- /dev/null
+++ b/src/tests/_internal/cli/services/test_endpoint_presets.py
@@ -0,0 +1,57 @@
+from pathlib import Path
+
+import pytest
+import yaml
+
+from dstack._internal.cli.services.endpoint_presets import EndpointPresetStore
+from dstack._internal.core.errors import CLIError
+from tests._internal.cli.endpoint_presets import get_endpoint_preset_recipe
+
+pytestmark = pytest.mark.windows
+
+
+class TestEndpointPresetStore:
+ def test_saves_and_lists_self_contained_recipe(self, tmp_path: Path):
+ store = EndpointPresetStore(tmp_path / "presets")
+ recipe = get_endpoint_preset_recipe()
+
+ path = store.save(recipe)
+
+ assert path == (tmp_path / "presets" / "models--Qwen--Qwen3.5-27B" / "8f3a12c4.yaml")
+ data = yaml.safe_load(path.read_text())
+ assert data["base"] == recipe.base
+ assert data["id"] == recipe.id
+ assert data["model"] == recipe.model
+ assert "recipes" not in data
+ assert store.list() == [recipe]
+ assert store.get(recipe.id) == recipe
+ assert not list(path.parent.glob("*.tmp"))
+
+ def test_replaces_same_recipe_id_atomically(self, tmp_path: Path):
+ store = EndpointPresetStore(tmp_path / "presets")
+ recipe = get_endpoint_preset_recipe()
+ store.save(recipe)
+
+ updated = recipe.copy(update={"context_length": 16384})
+ store.save(updated)
+
+ assert store.get(updated.id) == updated
+
+ def test_rejects_duplicate_recipe_id(self, tmp_path: Path):
+ store = EndpointPresetStore(tmp_path / "presets")
+ recipe = get_endpoint_preset_recipe()
+ store.save(recipe)
+ store.save(recipe.copy(update={"base": "Qwen/Another-Model"}))
+
+ with pytest.raises(CLIError, match="is not unique"):
+ store.get(recipe.id)
+
+ def test_rejects_recipe_without_successful_benchmark(self, tmp_path: Path):
+ store = EndpointPresetStore(tmp_path / "presets")
+ path = store.save(get_endpoint_preset_recipe())
+ data = yaml.safe_load(path.read_text())
+ data["validations"][0]["benchmarks"] = []
+ path.write_text(yaml.safe_dump(data, sort_keys=False))
+
+ with pytest.raises(CLIError, match="successful benchmark"):
+ store.list()
diff --git a/src/tests/_internal/core/models/test_endpoints.py b/src/tests/_internal/core/models/test_endpoints.py
new file mode 100644
index 0000000000..e0b54a8746
--- /dev/null
+++ b/src/tests/_internal/core/models/test_endpoints.py
@@ -0,0 +1,43 @@
+import pytest
+from pydantic import ValidationError
+
+from dstack._internal.core.models.endpoints import (
+ EndpointConfiguration,
+ EndpointModelBase,
+ EndpointModelRepo,
+)
+
+pytestmark = pytest.mark.windows
+
+
+class TestEndpointConfiguration:
+ def test_parses_string_as_exact_repo(self):
+ configuration = EndpointConfiguration(model="Qwen/Qwen3.5-27B")
+
+ assert isinstance(configuration.model, EndpointModelRepo)
+ assert configuration.model.exact_repo == "Qwen/Qwen3.5-27B"
+ assert configuration.model.api_model_name == "Qwen/Qwen3.5-27B"
+ assert not configuration.model.allows_variant_selection
+
+ def test_parses_base_model(self):
+ configuration = EndpointConfiguration(model={"base": "Qwen/Qwen3.5-27B"})
+
+ assert isinstance(configuration.model, EndpointModelBase)
+ assert configuration.model.exact_repo is None
+ assert configuration.model.api_model_name == "Qwen/Qwen3.5-27B"
+ assert configuration.model.allows_variant_selection
+
+ def test_parses_exact_repo_with_client_facing_name(self):
+ configuration = EndpointConfiguration(
+ model={
+ "repo": "community/Qwen3.5-27B-GPTQ-Int4",
+ "name": "Qwen/Qwen3.5-27B",
+ }
+ )
+
+ assert configuration.model.exact_repo == "community/Qwen3.5-27B-GPTQ-Int4"
+ assert configuration.model.api_model_name == "Qwen/Qwen3.5-27B"
+
+ def test_rejects_ambiguous_model_object(self):
+ with pytest.raises(ValidationError):
+ EndpointConfiguration(model={"base": "Qwen/base", "repo": "Qwen/repo"})
diff --git a/src/tests/_internal/utils/test_ssh.py b/src/tests/_internal/utils/test_ssh.py
index ad95936549..6dbca37860 100644
--- a/src/tests/_internal/utils/test_ssh.py
+++ b/src/tests/_internal/utils/test_ssh.py
@@ -1,8 +1,69 @@
import subprocess
import unittest
+from pathlib import Path
from unittest.mock import MagicMock, patch
-from dstack._internal.utils.ssh import check_required_ssh_version
+import pytest
+
+from dstack._internal.compat import IS_WINDOWS
+from dstack._internal.utils.path import FilePath
+from dstack._internal.utils.ssh import (
+ check_required_ssh_version,
+ include_ssh_config,
+ normalize_path,
+ update_ssh_config,
+)
+
+pytestmark = pytest.mark.windows
+
+
+class TestNormalizePath:
+ @pytest.mark.skipif(IS_WINDOWS, reason="POSIX OpenSSH home semantics")
+ def test_does_not_collapse_path_under_overridden_home(self, tmp_path, monkeypatch):
+ monkeypatch.setenv("HOME", str(tmp_path))
+ identity_file = tmp_path / ".dstack" / "ssh" / "key"
+
+ assert normalize_path(identity_file, collapse_user=True) == str(identity_file)
+
+ @pytest.mark.skipif(IS_WINDOWS, reason="POSIX OpenSSH home semantics")
+ def test_does_not_collapse_path_without_passwd_entry(self, tmp_path, monkeypatch):
+ monkeypatch.setattr("pwd.getpwuid", MagicMock(side_effect=KeyError))
+ identity_file = tmp_path / ".dstack" / "ssh" / "key"
+
+ assert normalize_path(identity_file, collapse_user=True) == str(identity_file)
+
+ def test_collapses_path_under_openssh_home(self):
+ identity_file = Path.home() / ".dstack" / "ssh" / "key"
+
+ assert normalize_path(identity_file, collapse_user=True) == "~/.dstack/ssh/key"
+
+
+@pytest.mark.skipif(IS_WINDOWS, reason="POSIX OpenSSH home semantics")
+class TestTemporaryHomeSSHConfig:
+ def test_writes_absolute_identity_file(self, tmp_path, monkeypatch):
+ home = tmp_path / "home"
+ monkeypatch.setenv("HOME", str(home))
+ identity_file = home / ".dstack" / "ssh" / "key"
+ config_file = home / ".dstack" / "ssh" / "config"
+
+ update_ssh_config(
+ config_file,
+ "test-run",
+ {"IdentityFile": FilePath(identity_file)},
+ )
+
+ assert f" IdentityFile {identity_file}\n" in config_file.read_text()
+
+ def test_writes_absolute_include(self, tmp_path, monkeypatch):
+ home = tmp_path / "home"
+ monkeypatch.setenv("HOME", str(home))
+ dstack_config = home / ".dstack" / "ssh" / "config"
+ user_config = home / ".ssh" / "config"
+ user_config.parent.mkdir(mode=0o700, parents=True)
+
+ include_ssh_config(dstack_config, user_config)
+
+ assert user_config.read_text() == f"Include {dstack_config}\n"
class TestCheckRequiredSSHVersion(unittest.TestCase):