[WIP] Preset patching and per-trial records

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

@@ -85,6 +85,7 @@
         "run_id": {"type": "string"},
         "run_name": {"type": "string"},
         "service_yaml": {"type": "string"},
+        "trial": {"type": "integer", "minimum": 1},
         "base": {"type": "string"},
         "model": {"type": "string"},
         "context_length": {"type": "integer", "minimum": 1},
@@ -101,6 +102,7 @@ class AgentFinalReport(CoreModel):
     run_id: Optional[uuid.UUID] = None
     run_name: Optional[str] = None
     service_yaml: Optional[str] = None
+    trial: Optional[PositiveInt] = None
     base: Optional[str] = None
     model: Optional[str] = None
     context_length: Optional[PositiveInt] = None
@@ -114,6 +116,7 @@ def validate_report(self) -> Self:
                 "run_id",
                 "run_name",
                 "service_yaml",
+                "trial",
                 "base",
                 "model",
                 "context_length",

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

@@ -132,6 +132,14 @@ class Preset(CoreModel):
     """Exact repo/path loaded by the service command."""
     context_length: PositiveInt
     """Token context length this preset was verified to support."""
+    trial: Optional[PositiveInt] = None
+    """Trial this preset was promoted from, within its creation session."""
+    min_context_length: Optional[PositiveInt] = None
+    """Context length asked for at creation. `context_length` may be below it: a
+    session that found no compliant trial verifies its best failed one."""
+    max_ttft: Optional[PositiveInt] = None
+    """Maximum p50 TTFT asked for at creation, in ms. The benchmark may exceed it,
+    for the same reason."""
     created_at: datetime
     service: ServiceConfiguration
     validations: list[PresetValidation]

src/dstack/_internal/cli/services/presets/agent.py

@@ -23,6 +23,7 @@
     print_preset_progress,
 )
 from dstack._internal.cli.services.presets.tail import (
+    _DirectoryMirror,
     _FileLineReader,
     _OffsetStore,
     _ProgressTailer,
@@ -415,20 +416,16 @@ async def _session_tailers(
             offset_key="runs",
             echo=agent_session.echo,
         ),
-        _RecordMirror(
-            source=workspace.trials_path,
-            target=agent_session.trials_path,
+        _DirectoryMirror(
+            source=workspace.trials_dir,
+            target=agent_session.trials_dir,
             redacted_values=redacted_values,
-            offset_store=offset_store,
-            offset_key="trials",
             echo=agent_session.echo,
         ),
-        _RecordMirror(
-            source=workspace.verifications_path,
-            target=agent_session.verifications_path,
+        _DirectoryMirror(
+            source=workspace.service_dir,
+            target=agent_session.service_dir,
             redacted_values=redacted_values,
-            offset_store=offset_store,
-            offset_key="verifications",
             echo=agent_session.echo,
         ),
     ]

src/dstack/_internal/cli/services/presets/apply.py

@@ -11,6 +11,7 @@
     format_preset_objective,
 )
 from dstack._internal.cli.services.presets.store import PresetStore
+from dstack._internal.cli.utils.common import warn
 from dstack._internal.core.errors import CLIError
 from dstack._internal.core.models.configurations import ServiceConfiguration
 from dstack._internal.core.models.profiles import ProfileParams
@@ -54,15 +55,19 @@ def apply_preset(
 
 
 def _validate_preset_matches(preset: Preset, *, configuration: PresetConfiguration) -> None:
-    """The referenced preset must serve what the configuration asks for."""
+    """The referenced preset must serve the model the configuration asks for.
+    A context length below the requested one warns instead of failing: the
+    preset is explicitly chosen by ID, and it may be the best a session could
+    verify (see `Preset.min_context_length`); the plan confirmation decides."""
     model_name = configuration.model.api_model_name
     service_model = preset.service.model
     if service_model is None or service_model.name.lower() != model_name.lower():
         raise CLIError(f"Preset {preset.id} does not serve {model_name}")
     if configuration.min_context_length is not None:
         if preset.context_length < configuration.min_context_length:
-            raise CLIError(
-                f"Preset {preset.id} does not support context length"
+            warn(
+                f"Preset {preset.id} is verified for context length"
+                f" {preset.context_length}, below the requested"
                 f" {configuration.min_context_length}"
             )
     if configuration.model.allows_variant_selection:

src/dstack/_internal/cli/services/presets/create.py

@@ -607,6 +607,8 @@ async def _create_preset(
             run=run,
             preset_configuration=source_configuration,
             report=report,
+            workspace_path=setup.workspace.path,
+            session_path=agent_session.path,
             preset_id=agent_session.preset_id or None,
             name=claimed_session_name(agent_session.read_manifest()),
         )

src/dstack/_internal/cli/services/presets/output.py

@@ -36,9 +36,11 @@ def _verifying(session: dict[str, Any]) -> bool:
 
 
 def _format_trial_spark(session: Optional[dict[str, Any]]) -> str:
-    """One glyph per trial, scaled within the run: the shape of the search.
-    A red `·` marks a trial that produced no benchmark at all; a yellow bar marks
-    one that measured but broke a constraint, since its number is real."""
+    """One glyph per trial, scaled from zero: bar heights compare as the
+    numbers do, so the size of a gain is visible. A red `·` marks a trial
+    that produced no benchmark at all, and a red bar one that measured but
+    broke a constraint. Gold marks the best result while no trial meets the
+    constraints; green takes over once one does."""
     if not isinstance(session, dict):
         return ""
     trials = session.get("trials")
@@ -51,40 +53,44 @@ def _format_trial_spark(session: Optional[dict[str, Any]]) -> str:
     values = [v for v in series if isinstance(v, (int, float))]
     if not values:
         return "·" * len(series)
-    low, high = min(values), max(values)
-    span = high - low
+    high = max(values)
+    passed = [v for v, f in zip(series, failed) if isinstance(v, (int, float)) and not f]
+    # The best trial is the answer the run found; everything else is context. Gold
+    # is that answer while none meets the constraints, so it gives way to green as
+    # soon as one does.
+    best = max(passed) if passed else max(values)
     out = []
     for value, is_failed in zip(series, failed):
         if not isinstance(value, (int, float)):
             out.append("[indian_red1]·[/]")
             continue
         glyph = (
             _SPARK_BLOCKS[-1]
-            if span <= 0
-            else _SPARK_BLOCKS[round((value - low) / span * (len(_SPARK_BLOCKS) - 1))]
+            if high <= 0
+            else _SPARK_BLOCKS[round(max(value, 0) / high * (len(_SPARK_BLOCKS) - 1))]
         )
-        # The best trial is the answer the run found; everything else is context.
-        # Yellow, not red: the trial measured, its number is real, and only the
-        # constraint breach makes it unusable. Red is reserved for `·`, where
-        # nothing came back at all.
         if is_failed:
-            style = "gold1"
+            style = "gold1" if not passed and value >= best else "indian_red1"
         else:
-            style = "bold sea_green3" if value >= high else "secondary"
+            style = "bold sea_green3" if value >= best else "secondary"
         out.append(f"[{style}]{glyph}[/]")
     return "".join(out)
 
 
-def _format_trial_progress(session: Optional[dict[str, Any]]) -> str:
+def _format_trial_progress(session: Optional[dict[str, Any]], *, in_flight: bool = False) -> str:
     """The ` (N/M)` suffix; stays outside the status markup to render in the
-    default color."""
+    default color. While trialing, `N` is the trial being worked on rather than
+    the completed count, so `trialing (2/3)` cannot read as two finished."""
     if not isinstance(session, dict):
         return ""
     trials = session.get("trials")
     trials_num = session.get("trials_num")
     if not isinstance(trials, dict) or not (trials.get("count") or isinstance(trials_num, int)):
         return ""
-    progress = str(trials.get("count") or 0)
+    count = trials.get("count") or 0
+    if in_flight:
+        count = min(count + 1, trials_num) if isinstance(trials_num, int) else count + 1
+    progress = str(count)
     if isinstance(trials_num, int):
         progress += f"/{trials_num}"
     return f" [secondary]({progress})[/]"
@@ -180,7 +186,9 @@ def _add_session(table: Table, session: dict[str, Any], *, verbose: bool = False
     status_key = str(session.get("status", ""))
     if status_key == "running" and _verifying(session):
         status_key = "verifying"
-    status = _format_status(status_key) + _format_trial_progress(session)
+    status = _format_status(status_key) + _format_trial_progress(
+        session, in_flight=status_key == "running"
+    )
     trials = session.get("trials")
     best = trials.get("best") if isinstance(trials, dict) else None
     # Nothing passed: fall back to the fastest attempt that did not.
@@ -279,8 +287,11 @@ def _add_preset(
         "": _format_trial_spark(creation),
         "CONSTRAINTS": format_preset_objective(
             preset,
-            min_context_length=(creation or {}).get("constraints", {}).get("min_context_length"),
-            max_ttft=(creation or {}).get("constraints", {}).get("max_ttft"),
+            # The preset carries what it was asked for; the creation record is the
+            # fallback for presets saved before it did.
+            min_context_length=preset.min_context_length
+            or (creation or {}).get("constraints", {}).get("min_context_length"),
+            max_ttft=preset.max_ttft or (creation or {}).get("constraints", {}).get("max_ttft"),
             verbose=verbose,
         ),
         "BENCHMARK": format_preset_benchmark(preset, verbose=verbose),
@@ -330,6 +341,18 @@ def format_preset_objective(
     return f"[secondary]{' '.join(parts)}[/]"
 
 
+def _breaches_constraints(preset: Preset) -> bool:
+    """Whether the verified benchmark misses what was asked for. A session that
+    found no compliant trial verifies its best failed one, so a preset can be
+    real, reproducible, and still fall short."""
+    metrics = preset.validations[0].benchmark.metrics
+    if preset.max_ttft is not None and metrics.ttft_ms.p50 > preset.max_ttft:
+        return True
+    return preset.min_context_length is not None and (
+        preset.context_length < preset.min_context_length
+    )
+
+
 def format_preset_benchmark(preset: Preset, *, verbose: bool = False) -> str:
     benchmark = preset.validations[0].benchmark
     metrics = benchmark.metrics
@@ -344,7 +367,12 @@ def format_preset_benchmark(preset: Preset, *, verbose: bool = False) -> str:
         f"ttft={_format_duration_ms(metrics.ttft_ms.p50)}",
         f"ctx={_format_token_count(preset.context_length)}",
     ]
-    return " ".join(parts)
+    text = " ".join(parts)
+    if _breaches_constraints(preset):
+        # Marked, not only dimmed: colour alone is not a signal. Same `*` a
+        # session row uses when it has nothing but failed trials to show.
+        return f"[secondary]*{text}[/]"
+    return text
 
 
 def _format_duration_ms(value: float) -> str:

src/dstack/_internal/cli/services/presets/presets.py

@@ -26,6 +26,9 @@ def build_preset(
     model: str,
     context_length: int,
     benchmark: PresetBenchmark,
+    trial: Optional[int] = None,
+    min_context_length: Optional[int] = None,
+    max_ttft: Optional[int] = None,
     preset_id: Optional[str] = None,
     name: Optional[str] = None,
 ) -> Preset:
@@ -45,6 +48,9 @@ def build_preset(
         id=preset_id or make_preset_id(service, context_length=context_length),
         model=model,
         context_length=context_length,
+        trial=trial,
+        min_context_length=min_context_length,
+        max_ttft=max_ttft,
         created_at=get_current_datetime(),
         service=service,
         validations=[validation],
@@ -73,6 +79,13 @@ def preset_to_data(preset: Preset) -> dict[str, Any]:
         **({"name": preset.name} if preset.name else {}),
         "model": preset.model,
         "context_length": preset.context_length,
+        **({"trial": preset.trial} if preset.trial is not None else {}),
+        **(
+            {"min_context_length": preset.min_context_length}
+            if preset.min_context_length is not None
+            else {}
+        ),
+        **({"max_ttft": preset.max_ttft} if preset.max_ttft is not None else {}),
         "created_at": preset.created_at.isoformat(),
         "service": service_configuration_to_preset_data(preset.service),
         "validations": [

src/dstack/_internal/cli/services/presets/prompt.py

@@ -18,7 +18,7 @@
 
 
 # TODO: reintroduce a `# Resume` section in system_prompt.md once session resume
-# (seeded from `runs.jsonl` and `trials.jsonl`) is designed.
+# (seeded from `runs.jsonl` and the trial records) is designed.
 def get_preset_agent_system_prompt(
     user_prompt: Optional[str] = None,
     baseline: bool = False,

src/dstack/_internal/cli/services/presets/redaction.py

@@ -4,6 +4,7 @@
 from typing import Any, Sequence
 
 _REDACTION = "[redacted]"
+_REDACTION_BYTES = _REDACTION.encode("utf-8")
 # Replacing shorter values such as "1" or "false" corrupts unrelated diagnostics.
 _MIN_REDACTED_SUBSTRING_LENGTH = 8
 _SENSITIVE_INHERITED_ENV_NAMES = (
@@ -47,6 +48,20 @@ def redact(value: str, redacted_values: Sequence[str]) -> str:
     return value
 
 
+def redact_bytes(value: bytes, redacted_values: Sequence[str]) -> bytes:
+    """As `redact`, on bytes. A copied file must keep its exact bytes, and
+    decoding it to text rewrites newlines and replaces non-UTF-8 bytes."""
+    for redacted_value in redacted_values:
+        # Environment values decode with surrogateescape, so encoding them back
+        # the same way is what returns their original bytes.
+        encoded = redacted_value.encode("utf-8", errors="surrogateescape")
+        if value == encoded:
+            return _REDACTION_BYTES
+        if len(redacted_value) >= _MIN_REDACTED_SUBSTRING_LENGTH:
+            value = value.replace(encoded, _REDACTION_BYTES)
+    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):