From 96eadb70c6395d5451c9c6bb30da96223e109c39 Mon Sep 17 00:00:00 2001 From: Andrew Klatzke Date: Tue, 14 Jul 2026 16:12:55 -0800 Subject: [PATCH 1/3] fix: fix for single-iteration optimizations --- .../optimization/src/ldai_optimizer/client.py | 76 +++++++++++++++---- 1 file changed, 62 insertions(+), 14 deletions(-) diff --git a/packages/optimization/src/ldai_optimizer/client.py b/packages/optimization/src/ldai_optimizer/client.py index db232b1f..532a8f88 100644 --- a/packages/optimization/src/ldai_optimizer/client.py +++ b/packages/optimization/src/ldai_optimizer/client.py @@ -15,6 +15,7 @@ about the caller's broader runtime environment beyond the key itself. """ +import asyncio import dataclasses import json import logging @@ -2118,6 +2119,15 @@ def _persist_and_forward( api_client.patch_agent_optimization_result( project_key, optimization_key, result_id, patch ) + # When the winning result is marked successful, make sure + # _last_optimization_result_id tracks it. Without this the + # auto-commit PATCH (which attaches createdVariationKey) is + # sent to the last-posted Phase 2 record rather than to the + # winner, giving a non-winning RUNNING record the latest + # updatedAt and causing the backend to report the run as still + # running even after the optimization completes. + if status == "success": + self._last_optimization_result_id = result_id # Reset tracking state after terminal events so the next main-loop # attempt starts fresh. @@ -2673,16 +2683,20 @@ async def _run_cost_latency_phase( frozen_user_input = winning_ctx.user_input # Build a deterministic, deduplicated list of models to evaluate: - # start with the Phase 1 winner's model, then add each model_choice - # that hasn't been seen yet. This guarantees every user-selected model - # is tried exactly once, in a predictable order. + # always start from model_choices, skipping the Phase 1 winner so it + # doesn't appear as an extra input inside the quality iteration in the + # UI. Fall back to the Phase 1 winner only when no distinct choices + # are provided. phase1_model = winning_ctx.current_model or "" seen_models: set = {phase1_model} - ordered_models: List[str] = [phase1_model] + ordered_models: List[str] = [] for m in self._options.model_choices or []: if m not in seen_models: seen_models.add(m) ordered_models.append(m) + # Fall back to Phase 1 model only if no distinct alternatives exist. + if not ordered_models: + ordered_models.append(phase1_model) # Ensure at least 2 iterations while len(ordered_models) < 2: ordered_models.append(ordered_models[-1]) @@ -2714,12 +2728,35 @@ async def _run_cost_latency_phase( variables=frozen_variables, user_input=frozen_user_input, ) + # Pre-populate placeholder gate score keys so the "generating" + # status update carries the same score keys as the final Phase 2 + # result. Without these placeholders the UI bucketing logic + # (which groups by gate-key presence) cannot distinguish this + # record from a Phase 1 result and places it in the wrong + # iteration group. The placeholders are overwritten with real + # values after the agent call and gate evaluation complete. + gate_placeholders: Dict[str, JudgeResult] = {} + if self._options.latency_optimization: + gate_placeholders["_latency_gate"] = JudgeResult( + score=0.0, rationale="evaluating" + ) + if self._options.token_optimization: + gate_placeholders["_cost_gate"] = JudgeResult( + score=0.0, rationale="evaluating" + ) + if gate_placeholders: + ctx = dataclasses.replace( + ctx, scores={**ctx.scores, **gate_placeholders} + ) self._safe_status_update("generating", ctx, iteration) try: - ctx = await self._execute_agent_turn(ctx, iteration) - except Exception: + ctx = await asyncio.wait_for( + self._execute_agent_turn(ctx, iteration), + timeout=120, + ) + except (Exception, asyncio.TimeoutError): logger.warning( - "[Phase 2 Iter %d] -> Agent call failed (model=%s); " + "[Phase 2 Iter %d] -> Agent call failed or timed out (model=%s); " "skipping this model and trying the next", iteration, self._current_model, @@ -2758,16 +2795,27 @@ async def _run_cost_latency_phase( if i < max_iters - 1: self._safe_status_update("turn completed", ctx, iteration) - # Send terminal FAILED status for each non-winning model attempt. - # We use _safe_status_update directly rather than _handle_failure so that - # exploratory Phase 2 misses don't corrupt _last_run_succeeded, - # _last_succeeded_context, or trigger on_failing_result — those are - # run-level signals that should only fire if the whole optimization fails. - for failed_ctx in non_candidates: - self._safe_status_update("failure", failed_ctx, failed_ctx.iteration) + # Phase 2 is complete. The `status` field on each result carries the + # *run-level* outcome (PASSED / FAILED / RUNNING), not the quality of + # that individual result — the scores already encode individual quality. + # The backend derives the visible run status from the highest-iteration + # result, so every Phase 2 result must end with status=PASSED after a + # successful run; otherwise the highest-numbered result keeps the run in + # RUNNING indefinitely. We use _safe_status_update directly (not + # _handle_failure / _handle_success) for non-winners so that + # _last_run_succeeded, _last_succeeded_context, and on_failing_result are + # not corrupted — those are reserved for run-level outcomes. + for non_candidate_ctx in non_candidates: + self._safe_status_update("success", non_candidate_ctx, non_candidate_ctx.iteration) if candidates: best = self._pick_best_candidate(candidates) + # Non-best candidates: mark PASSED (run succeeded) before the winner + # so _handle_success remains the very last update, preserving the + # correct _last_succeeded_context for the caller. + for other in candidates: + if other.iteration != best.iteration: + self._safe_status_update("success", other, other.iteration) # Suppress on_passing_result here — the caller fires it once with the # true final winner after Phase 2 returns, so it is never double-fired. self._handle_success(best, best.iteration, suppress_user_callbacks=True) From eba89fcb84023753a696e5f5a974d30fce822814 Mon Sep 17 00:00:00 2001 From: Andrew Klatzke Date: Tue, 14 Jul 2026 16:15:16 -0800 Subject: [PATCH 2/3] chore: remove unnecessary comment --- packages/optimization/src/ldai_optimizer/client.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/packages/optimization/src/ldai_optimizer/client.py b/packages/optimization/src/ldai_optimizer/client.py index 532a8f88..ff10d8a6 100644 --- a/packages/optimization/src/ldai_optimizer/client.py +++ b/packages/optimization/src/ldai_optimizer/client.py @@ -2728,13 +2728,7 @@ async def _run_cost_latency_phase( variables=frozen_variables, user_input=frozen_user_input, ) - # Pre-populate placeholder gate score keys so the "generating" - # status update carries the same score keys as the final Phase 2 - # result. Without these placeholders the UI bucketing logic - # (which groups by gate-key presence) cannot distinguish this - # record from a Phase 1 result and places it in the wrong - # iteration group. The placeholders are overwritten with real - # values after the agent call and gate evaluation complete. + gate_placeholders: Dict[str, JudgeResult] = {} if self._options.latency_optimization: gate_placeholders["_latency_gate"] = JudgeResult( From 3c65a6e4651e910e3547cc7114f785793114e911 Mon Sep 17 00:00:00 2001 From: Andrew Klatzke Date: Wed, 15 Jul 2026 11:55:50 -0800 Subject: [PATCH 3/3] fix: move failure forward --- .../optimization/src/ldai_optimizer/client.py | 89 ++++++++++++++++--- .../src/ldai_optimizer/ld_api_client.py | 44 +++++++-- 2 files changed, 116 insertions(+), 17 deletions(-) diff --git a/packages/optimization/src/ldai_optimizer/client.py b/packages/optimization/src/ldai_optimizer/client.py index ff10d8a6..01ea6883 100644 --- a/packages/optimization/src/ldai_optimizer/client.py +++ b/packages/optimization/src/ldai_optimizer/client.py @@ -49,6 +49,7 @@ AgentOptimizationResultPatch, AgentOptimizationResultPost, LDApiClient, + LDApiError, ) from ldai_optimizer.prompts import ( build_message_history_text, @@ -340,7 +341,7 @@ def _build_agent_config_for_context( create_tracker=self._agent_config.create_tracker, evaluator=self._agent_config.evaluator, model=ModelConfig( - name=ctx.current_model or "", + name=_strip_provider_prefix(ctx.current_model or ""), parameters=ctx.current_parameters, ), instructions=instructions, @@ -1094,7 +1095,10 @@ async def _get_agent_config( ) agent_config = dataclasses.replace( agent_config, - model=ModelConfig(name=model_config_key, parameters=model_parameters or {}), + model=ModelConfig( + name=_strip_provider_prefix(model_config_key), + parameters=model_parameters or {}, + ), ) else: # variation() returns the raw JSON before chevron.render(), so instructions @@ -1880,7 +1884,21 @@ async def optimize_from_config( api_client=api_client, ) - optimization_options = self._build_options_from_config( + # Preflight: verify that the API key has write access to the results endpoint + # before any agent work begins. A read-only or wrong-project key would + # otherwise silently 403 on every result POST mid-run, leaving the Results + # tab empty with no visible error. + try: + api_client.verify_write_access(options.project_key, optimization_key) + except LDApiError as exc: + raise ValueError( + f"API key does not have write access to project '{options.project_key}' " + f"(optimization: '{optimization_key}'). Verify that your LAUNCHDARKLY_API_KEY " + f"has the 'Writer' role (or equivalent) for this project before running an " + f"optimization. Original error: {exc}" + ) from exc + + optimization_options, _log_persist_summary = self._build_options_from_config( config, options, api_client, optimization_key, run_id, model_configs ) if isinstance(optimization_options, GroundTruthOptimizationOptions): @@ -1888,6 +1906,8 @@ async def optimize_from_config( else: result = await self._run_optimization(agent_config, optimization_options) + _log_persist_summary() + if (optimization_options.auto_commit and options.auto_commit and self._last_run_succeeded and self._last_succeeded_context): created_key = self._commit_variation( @@ -1915,7 +1935,7 @@ def _build_options_from_config( optimization_key: str, run_id: str, model_configs: Optional[List[Dict[str, Any]]] = None, - ) -> "Union[OptimizationOptions, GroundTruthOptimizationOptions]": + ) -> "Tuple[Union[OptimizationOptions, GroundTruthOptimizationOptions], Any]": """Map a fetched AgentOptimization config + user options into the appropriate options type. When the config contains groundTruthResponses, the three lists (groundTruthResponses, @@ -1934,7 +1954,8 @@ def _build_options_from_config( :param optimization_key: String key of the parent agent_optimization record. :param run_id: UUID that groups all result records for this run. :param model_configs: Pre-fetched list of model config dicts for resolving modelConfigKey. - :return: OptimizationOptions or GroundTruthOptimizationOptions. + :return: Tuple of (OptimizationOptions or GroundTruthOptimizationOptions, summary_fn). + Call summary_fn() after the optimization loop to log persistence health. """ judges: Dict[str, OptimizationJudge] = {} @@ -1967,6 +1988,18 @@ def _build_options_from_config( config_version: int = config["version"] _cached_model_configs: List[Dict[str, Any]] = list(model_configs or []) + # Warn early about model choices that have no matching model config — these + # are likely retired or misspelled and will fail at run time with an opaque + # provider error rather than a useful message. + if _cached_model_configs: + for raw_model in config.get("modelChoices") or []: + if not _find_model_config(raw_model, _cached_model_configs): + logger.warning( + "Model choice '%s' was not found in the project's model configs — " + "it may be retired or misspelled and will likely fail at run time.", + raw_model, + ) + # Maps logical iteration number → result record id. Each new main-loop # iteration (plus the init iteration 0) POSTs a fresh record; subsequent # status events for that same iteration PATCH the existing record. @@ -1985,6 +2018,13 @@ def _build_options_from_config( # init iteration 0, or non-final GT samples) are left in a stale state. _last_open_iteration: int = -1 + # Persistence health counters — incremented on every failed POST or PATCH + # so a single summary warning can be emitted at run end. + _post_failures: int = 0 + _patch_failures: int = 0 + _post_successes: int = 0 + _patch_successes: int = 0 + def _resolve_model_config_key(model_name: str) -> str: if not model_name: return "" @@ -2006,6 +2046,7 @@ def _persist_and_forward( ctx: OptimizationContext, ) -> None: nonlocal _in_validation_phase, _validation_parent_iteration, _last_open_iteration + nonlocal _post_failures, _patch_failures, _post_successes, _patch_successes # _safe_status_update (the caller) already wraps this entire function in # a try/except, so errors here are caught and logged without aborting the run. mapped = _OPTIMIZATION_STATUS_MAP.get( @@ -2052,11 +2093,10 @@ def _persist_and_forward( "agentOptimizationVersion": config_version, "iteration": logical_iteration, "instructions": snapshot.current_instructions, + "userInput": snapshot.user_input or "", } if snapshot.current_parameters: post_payload["parameters"] = snapshot.current_parameters - if snapshot.user_input: - post_payload["userInput"] = snapshot.user_input result_id = api_client.post_agent_optimization_result( project_key, optimization_key, post_payload ) @@ -2064,6 +2104,9 @@ def _persist_and_forward( _iteration_result_ids[logical_iteration] = result_id self._last_optimization_result_id = result_id _last_open_iteration = logical_iteration + _post_successes += 1 + else: + _post_failures += 1 # Phase 2: PATCH the record with current status and available telemetry. result_id = _iteration_result_ids.get(logical_iteration) @@ -2116,9 +2159,12 @@ def _persist_and_forward( "parameters": snapshot.current_parameters, "modelConfigKey": _resolve_model_config_key(snapshot.current_model or ""), } - api_client.patch_agent_optimization_result( + if api_client.patch_agent_optimization_result( project_key, optimization_key, result_id, patch - ) + ): + _patch_successes += 1 + else: + _patch_failures += 1 # When the winning result is marked successful, make sure # _last_optimization_result_id tracks it. Without this the # auto-commit PATCH (which attaches createdVariationKey) is @@ -2142,6 +2188,27 @@ def _persist_and_forward( except Exception: logger.exception("User on_status_update callback failed for status=%s", status) + def _log_persist_summary() -> None: + """Emit a single summary line about result-persistence health at run end.""" + total_posts = _post_successes + _post_failures + total_patches = _patch_successes + _patch_failures + if _post_failures or _patch_failures: + logger.warning( + "Result persistence summary: %d/%d POSTs and %d/%d PATCHes to the " + "LaunchDarkly API failed — the Results tab may be incomplete. " + "Check your API key and network connectivity.", + _post_successes, + total_posts, + _patch_successes, + total_patches, + ) + else: + logger.debug( + "Result persistence summary: all %d POSTs and %d PATCHes succeeded.", + total_posts, + total_patches, + ) + # If we have ground truth responses, we provide a different # configuration options type that contains the bundled GroundTruthSamples # so that the ultimate output is correctly formatted. @@ -2183,7 +2250,7 @@ def _persist_and_forward( latency_optimization=config.get("latencyOptimization"), token_optimization=config.get("tokenOptimization"), auto_commit=config.get("autoCommit", True), - ) + ), _log_persist_summary variable_choices: List[Dict[str, Any]] = config["variableChoices"] or [{}] user_input_options: Optional[List[str]] = config["userInputOptions"] or None @@ -2206,7 +2273,7 @@ def _persist_and_forward( latency_optimization=config.get("latencyOptimization"), token_optimization=config.get("tokenOptimization"), auto_commit=config.get("autoCommit", True), - ) + ), _log_persist_summary async def _execute_agent_turn( self, diff --git a/packages/optimization/src/ldai_optimizer/ld_api_client.py b/packages/optimization/src/ldai_optimizer/ld_api_client.py index db3a0976..6dbd580a 100644 --- a/packages/optimization/src/ldai_optimizer/ld_api_client.py +++ b/packages/optimization/src/ldai_optimizer/ld_api_client.py @@ -105,12 +105,12 @@ class _AgentOptimizationResultPostRequired(TypedDict): agentOptimizationVersion: int iteration: int instructions: str + userInput: str class AgentOptimizationResultPost(_AgentOptimizationResultPostRequired, total=False): """Payload for POST /agent-optimizations/{key}/results — creates a new result record.""" - userInput: str parameters: Dict[str, Any] @@ -343,6 +343,34 @@ def get_agent_optimization( raw = self._request("GET", path) return _parse_agent_optimization(raw) + def verify_write_access(self, project_key: str, optimization_key: str) -> None: + """Probe-POST to verify write access to the results endpoint before a run starts. + + Sends an intentionally invalid payload (``iteration: -1``) so the API will + reject the body with a 400 if auth passes — confirming write access without + creating a permanent record. Only 401 and 403 responses are re-raised; + a 400 (bad input) means authentication and authorisation both succeeded. + + :param project_key: LaunchDarkly project key. + :param optimization_key: Key of the agent optimization config. + :raises LDApiError: If the API returns 401 or 403. + """ + path = f"/api/v2/projects/{project_key}/agent-optimizations/{optimization_key}/results" + probe: AgentOptimizationResultPost = { + "runId": "__preflight__", + "agentOptimizationVersion": -1, + "iteration": -1, + "instructions": "", + "userInput": "", + } + try: + self._request("POST", path, body=probe) + except LDApiError as exc: + if exc.status_code in (401, 403): + raise + # 400 (invalid body) or any other non-auth error means we reached + # the API and it accepted our credentials — write access is confirmed. + def post_agent_optimization_result( self, project_key: str, optimization_key: str, payload: AgentOptimizationResultPost ) -> Optional[str]: @@ -361,7 +389,7 @@ def post_agent_optimization_result( result = self._request("POST", path, body=payload) return result.get("id") if isinstance(result, dict) else None except LDApiError as exc: - logger.debug( + logger.warning( "Failed to persist optimization result (optimization_key=%s, iteration=%s): %s", optimization_key, payload.get("iteration"), @@ -369,7 +397,7 @@ def post_agent_optimization_result( ) return None except Exception as exc: - logger.debug( + logger.warning( "Unexpected error persisting optimization result (optimization_key=%s, iteration=%s): %s", optimization_key, payload.get("iteration"), @@ -379,7 +407,7 @@ def post_agent_optimization_result( def patch_agent_optimization_result( self, project_key: str, optimization_key: str, result_id: str, payload: AgentOptimizationResultPatch - ) -> None: + ) -> bool: """Update an existing iteration result record. Errors are caught and logged rather than raised so that persistence @@ -389,19 +417,23 @@ def patch_agent_optimization_result( :param optimization_key: String key of the parent agent_optimization record. :param result_id: ID of the result record to update. :param payload: PATCH payload with fields to update. + :return: True if the update succeeded, False on any error. """ path = f"/api/v2/projects/{project_key}/agent-optimizations/{optimization_key}/results/{result_id}" try: self._request("PATCH", path, body=payload) + return True except LDApiError as exc: - logger.debug( + logger.warning( "Failed to update optimization result (result_id=%s): %s", result_id, exc, ) + return False except Exception as exc: - logger.debug( + logger.warning( "Unexpected error updating optimization result (result_id=%s): %s", result_id, exc, ) + return False