Apply preset review fixes: packaging, resilience, dead code, test trims

Fix the wheel force-include paths left pointing at the removed endpoints
package, so installed CLIs find the bundled skills, and drop stale endpoint
terminology from the agent system prompt.

Robustness: a corrupt preset file no longer breaks list/find and stays
deletable through the CLI (its warning goes to a new stderr console so --json
stdout stays parseable); session manifests and offsets are written atomically
(tmp + fsync + replace, with a Windows retry for concurrent readers); a
read-only logs -f exits with a re-follow hint when the owner and agent are
both gone instead of polling forever; follow always takes the finalize claim;
a failed cleanup no longer retries without keep_final_service.

Dead code: the write-only session timestamp field, the no-op stderr report
merge, the unused PresetTrial model, the _LineStream protocol, and dead
defensive branches. Deduplicate: structure redaction (now also redacting dict
keys in report scrubbing), EnvSentinel resolution, run-name parsing, session
close + workspace removal, and the no-fleets error.

Tests: hoist repeated scaffolding (fake claude, patch blocks, plain console,
session dirs), parametrize near-duplicates, merge overlapping tests, move
model and output tests next to what they test, and cover the corrupt-preset
CLI paths.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: Andrey Cheptsov

pyproject.toml

@@ -73,8 +73,8 @@ artifacts = [
 ]
 
 [tool.hatch.build.targets.wheel.force-include]
-"skills/dstack/SKILL.md" = "dstack/_internal/cli/services/endpoints/resources/skills/dstack/SKILL.md"
-"skills/dstack-prototyping/SKILL.md" = "dstack/_internal/cli/services/endpoints/resources/skills/dstack-prototyping/SKILL.md"
+"skills/dstack/SKILL.md" = "dstack/_internal/cli/services/presets/resources/skills/dstack/SKILL.md"
+"skills/dstack-prototyping/SKILL.md" = "dstack/_internal/cli/services/presets/resources/skills/dstack-prototyping/SKILL.md"
 
 [tool.hatch.metadata.hooks.fancy-pypi-readme]
 content-type = "text/markdown"

src/dstack/_internal/cli/commands/preset.py

@@ -38,7 +38,7 @@
     load_profile_from_args,
     register_profile_args,
 )
-from dstack._internal.cli.utils.common import confirm_ask, console
+from dstack._internal.cli.utils.common import confirm_ask, console, warn
 from dstack._internal.core.errors import CLIError, ConfigurationError
 from dstack._internal.core.models.profiles import ProfileParams
 from dstack._internal.core.services import is_valid_dstack_resource_name
@@ -216,14 +216,14 @@ def _reconcile(self) -> None:
             reconcile_detached_sessions(PresetStore())
 
     def _list(self, args: argparse.Namespace) -> None:
-        base = getattr(args, "base", None)
-        repo = getattr(args, "repo", None)
-        if getattr(args, "json", False):
+        base = args.base
+        repo = args.repo
+        if args.json:
             self._reconcile()
             presets = _filter_presets(PresetStore().list(), base=base, repo=repo)
             print(PresetListOutput(presets=presets).json())
             return
-        verbose = getattr(args, "verbose", False)
+        verbose = args.verbose
         while True:
             self._reconcile()
             presets = PresetStore().list()
@@ -334,35 +334,33 @@ def _apply(self, args: argparse.Namespace) -> None:
     def _delete(self, args: argparse.Namespace) -> None:
         store = PresetStore()
         if args.preset is not None:
-            preset = store.get(args.preset) or store.find_by_name(args.preset)
-            if preset is None:
-                raise CLIError(f"Preset {args.preset!r} does not exist")
-            presets = [preset]
-            message = f"Delete preset [code]{preset.id}[/] for [code]{preset.base}[/]?"
+            try:
+                preset = store.get(args.preset) or store.find_by_name(args.preset)
+            except CLIError as e:
+                # A corrupt preset file must still be removable by ID.
+                warn(str(e), stderr=True)
+                preset_ids = [args.preset]
+                description = f"preset [code]{args.preset}[/]"
+            else:
+                if preset is None:
+                    raise CLIError(f"Preset {args.preset!r} does not exist")
+                preset_ids = [preset.id]
+                description = f"preset [code]{preset.id}[/] for [code]{preset.base}[/]"
         else:
             target = args.base or args.repo
             presets = _filter_presets(store.list(), base=args.base, repo=args.repo)
             if not presets:
                 kind = "base model" if args.base else "model repo"
                 raise CLIError(f"No presets found for {kind} {target!r}")
-            message = (
-                f"Delete {len(presets)} preset"
-                f"{'s' if len(presets) != 1 else ''} for [code]{target}[/]?"
-            )
-        if not args.yes and not confirm_ask(message):
+            preset_ids = [preset.id for preset in presets]
+            count = f"{len(presets)} preset{'s' if len(presets) != 1 else ''}"
+            description = f"{count} for [code]{target}[/]"
+        if not args.yes and not confirm_ask(f"Delete {description}?"):
             console.print("\nExiting...")
             return
-        for preset in presets:
-            store.delete(preset.id)
-        if args.preset is not None:
-            console.print(
-                f"Preset [code]{presets[0].id}[/] for [code]{presets[0].base}[/] deleted"
-            )
-        else:
-            console.print(
-                f"Deleted {len(presets)} preset{'s' if len(presets) != 1 else ''} "
-                f"for [code]{args.base or args.repo}[/]"
-            )
+        for preset_id in preset_ids:
+            store.delete(preset_id)
+        console.print(f"Deleted {description}")
 
 
 def _add_configuration_args(parser: argparse.ArgumentParser) -> None:

src/dstack/_internal/cli/models/preset_trials.py

@@ -12,7 +12,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Any, AsyncIterator, Callable, Iterator, Optional, Protocol, Sequence
+from typing import Any, AsyncIterator, Callable, Iterator, Optional, Sequence
 
 import psutil
 import yaml
@@ -40,6 +40,8 @@
 _USER_PROMPT_FILENAME = "user_prompt.md"
 _PROGRESS_ENV = "DSTACK_PRESET_PROGRESS_LOG"
 _REDACTION = "[redacted]"
+# Replacing shorter values such as "1" or "false" corrupts unrelated diagnostics.
+_MIN_REDACTED_SUBSTRING_LENGTH = 8
 _CLAUDE_TOOLS = "Bash,Read,Write,Edit,WebFetch,WebSearch,StructuredOutput"
 _CLAUDE_EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")
 _RESUME_DELAYS_SECONDS: tuple[int, ...] = (30, 60, 120)
@@ -137,7 +139,6 @@ def agent_stderr_path(self) -> Path:
 @dataclass
 class PresetAgentSession:
     path: Path
-    timestamp: str
     debug: bool
     preset_id: str = ""
     # Whether progress lines echo to this process's console (a live attach), on
@@ -328,7 +329,6 @@ def create_preset_agent_session(
 ) -> PresetAgentSession:
     if configuration.name is None:
         raise CLIError("The service name is required to save agent output")
-    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S-%fZ")
     parent = get_presets_dir()
     path: Optional[Path] = None
     try:
@@ -371,7 +371,7 @@ def create_preset_agent_session(
             shutil.rmtree(path, ignore_errors=True)
         raise CLIError(f"Could not create agent output under {parent}: {e}") from e
     assert path is not None
-    return PresetAgentSession(path=path, timestamp=timestamp, debug=debug, preset_id=preset_id)
+    return PresetAgentSession(path=path, debug=debug, preset_id=preset_id)
 
 
 def _get_claude_version(auth: "ClaudeAuth") -> Optional[str]:
@@ -407,7 +407,7 @@ def _get_claude_auth_status(auth: "ClaudeAuth") -> dict[str, Any]:
 
 def load_resumable_agent_session(preset_id: str) -> PresetAgentSession:
     path = get_presets_dir() / preset_id
-    session = PresetAgentSession(path=path, timestamp="", debug=False, preset_id=preset_id)
+    session = PresetAgentSession(path=path, debug=False, preset_id=preset_id)
     manifest = session.read_manifest()
     if not path.is_dir() or not manifest:
         raise CLIError(f"Unknown preset: {preset_id}")
@@ -424,7 +424,6 @@ def load_resumable_agent_session(preset_id: str) -> PresetAgentSession:
     if not manifest.get("claude_session_id"):
         raise CLIError(f"Preset {preset_id} creation stopped before it started; create a new one")
     session.debug = bool(manifest.get("debug"))
-    session.timestamp = str(manifest.get("created_at") or "")
     return session
 
 
@@ -455,7 +454,7 @@ def session_process_alive(manifest: dict[str, Any]) -> bool:
 
 def load_attachable_agent_session(preset_id: str) -> PresetAgentSession:
     path = get_presets_dir() / preset_id
-    session = PresetAgentSession(path=path, timestamp="", debug=False, preset_id=preset_id)
+    session = PresetAgentSession(path=path, debug=False, preset_id=preset_id)
     manifest = session.read_manifest()
     if not path.is_dir() or not manifest:
         raise CLIError(f"Unknown preset: {preset_id}")
@@ -481,14 +480,13 @@ def load_attachable_agent_session(preset_id: str) -> PresetAgentSession:
             f" stop or detach it there with Ctrl+C"
         )
     session.debug = bool(manifest.get("debug"))
-    session.timestamp = str(manifest.get("created_at") or "")
     return session
 
 
 def load_agent_session(preset_id: str) -> PresetAgentSession:
     """Loads a session of any status for read-only inspection (its log)."""
     path = get_presets_dir() / preset_id
-    session = PresetAgentSession(path=path, timestamp="", debug=False, preset_id=preset_id)
+    session = PresetAgentSession(path=path, debug=False, preset_id=preset_id)
     if not path.is_dir() or not session.read_manifest():
         raise CLIError(f"Unknown preset: {preset_id}")
     return session
@@ -601,7 +599,7 @@ def iter_agent_sessions() -> Iterator[PresetAgentSession]:
         return
     for path in sorted(root.iterdir()):
         if path.is_dir() and not path.name.startswith((".", "models--")):
-            yield PresetAgentSession(path=path, timestamp="", debug=False, preset_id=path.name)
+            yield PresetAgentSession(path=path, debug=False, preset_id=path.name)
 
 
 def find_session_name_claims(name: str) -> list[PresetAgentSession]:
@@ -882,7 +880,8 @@ def get_redacted_values(values: Sequence[str]) -> tuple[str, ...]:
 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)
+            value == redacted
+            or (len(redacted) >= _MIN_REDACTED_SUBSTRING_LENGTH and redacted in value)
             for redacted in redacted_values
         )
     if isinstance(value, dict):
@@ -900,12 +899,25 @@ def redact(value: str, redacted_values: Sequence[str]) -> str:
     for redacted_value in redacted_values:
         if value == redacted_value:
             return _REDACTION
-        # Replacing short values such as "1" or "false" corrupts unrelated diagnostics.
-        if len(redacted_value) >= 8:
+        if len(redacted_value) >= _MIN_REDACTED_SUBSTRING_LENGTH:
             value = value.replace(redacted_value, _REDACTION)
     return value
 
 
+def redact_structure(value: Any, redacted_values: Sequence[str]) -> Any:
+    """Recursively redacts every string (including dict keys) in a JSON-like value."""
+    if isinstance(value, str):
+        return redact(value, redacted_values)
+    if isinstance(value, list):
+        return [redact_structure(item, redacted_values) for item in value]
+    if isinstance(value, dict):
+        return {
+            redact(key, redacted_values): redact_structure(item, redacted_values)
+            for key, item in value.items()
+        }
+    return value
+
+
 def _validate_control_socket_path(build_root: Path) -> None:
     if IS_WINDOWS:
         return
@@ -1080,9 +1092,31 @@ def _prepare_subprocess_command(command: list[str]) -> list[str]:
 
 
 def _write_private_text(path: Path, content: str) -> None:
-    path.write_text(content, encoding="utf-8")
-    if not IS_WINDOWS:
-        path.chmod(0o600)
+    # Atomic tmp + fsync + replace (mkstemp already creates the file 0600), so
+    # a crash mid-write cannot leave a truncated manifest or offsets file.
+    fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=f".{path.name}.", suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as f:
+            f.write(content)
+            f.flush()
+            os.fsync(f.fileno())
+        try:
+            os.replace(temporary, path)
+        except PermissionError:
+            if not IS_WINDOWS:
+                raise
+            # A concurrent reader (a viewer polling the manifest) can hold the
+            # destination open without FILE_SHARE_DELETE; retry briefly, then
+            # prefer an in-place write over crashing the owner.
+            for _ in range(3):
+                time.sleep(0.01)
+                with suppress(PermissionError):
+                    os.replace(temporary, path)
+                    return
+            path.write_text(content, encoding="utf-8")
+    finally:
+        with suppress(FileNotFoundError):
+            os.unlink(temporary)
 
 
 def _write_debug_trace(
@@ -1096,7 +1130,7 @@ def _write_debug_trace(
         datetime.now(timezone.utc).isoformat(timespec="microseconds").replace("+00:00", "Z")
     )
     try:
-        event = _redact_trace_value(json.loads(text), redacted_values)
+        event = redact_structure(json.loads(text), redacted_values)
         record = {"timestamp": timestamp, "stream": stream_name, "event": event}
     except json.JSONDecodeError:
         record = {
@@ -1110,19 +1144,6 @@ def _write_debug_trace(
         f.flush()
 
 
-def _redact_trace_value(value: Any, redacted_values: Sequence[str]) -> Any:
-    if isinstance(value, str):
-        return redact(value, redacted_values)
-    if isinstance(value, list):
-        return [_redact_trace_value(item, redacted_values) for item in value]
-    if isinstance(value, dict):
-        return {
-            redact(key, redacted_values): _redact_trace_value(item, redacted_values)
-            for key, item in value.items()
-        }
-    return value
-
-
 @asynccontextmanager
 async def _session_tailers(
     *,
@@ -1182,7 +1203,7 @@ async def _collect_agent_output(
     """Parses the agent's stream files until it exits; safe alongside a live
     process or over the remains of a finished one."""
     offset_store = _OffsetStore(agent_session.path / ".offsets.json")
-    stdout_output, stderr_output = await asyncio.gather(
+    stdout_output, _ = await asyncio.gather(
         _read_process_stream(
             stream=_FileLineReader(
                 workspace.agent_stdout_path,
@@ -1208,12 +1229,9 @@ async def _collect_agent_output(
             agent_session=agent_session,
         ),
     )
-    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
-    return output
+    # stderr is tailed with parse_result=False — it feeds the debug trace and
+    # advances the persisted offset, but can never contribute report data.
+    return stdout_output
 
 
 async def attach_preset_agent(
@@ -1242,7 +1260,7 @@ def agent_alive() -> bool:
 
 async def _read_process_stream(
     *,
-    stream: "_LineStream",
+    stream: "_FileLineReader",
     stream_name: str,
     parse_result: bool,
     redacted_values: Sequence[str],
@@ -1370,10 +1388,6 @@ def _terminate_windows_process_tree(pid: int) -> None:
     psutil.wait_procs(alive, timeout=3)
 
 
-class _LineStream(Protocol):
-    async def readline(self) -> bytes: ...
-
-
 class _FileLineReader:
     """`readline()` over a growing file, so stream parsing survives CLI
     restarts: offsets persist, and a later attach continues exactly where the