From 2d0c477f1d0fe0c193026d8ad54c11a72de9b3aa Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Mon, 9 Mar 2026 12:36:38 +0100 Subject: [PATCH 01/12] Add file-based usage metrics for local runs (#110) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Write per-LLM-call metrics (model, tokens, duration, success/failure) to usage_metrics.jsonl in the run output directory. Works without a database — designed for local/offline runs where the DB-backed token_metrics_store is unavailable. - Add USAGE_METRICS_JSONL to ExtraFilenameEnum - New usage_metrics.py module with set/get path and record function - Extend LLMExecutor._record_attempt_token_metrics() to also write file-based metrics alongside existing DB recording - Wire usage metrics path in ExecutePipeline.run() - Add usage_metrics.jsonl to progress ignore list Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_api/filenames.py | 1 + .../llm_util/llm_executor.py | 19 +++++ .../llm_util/usage_metrics.py | 77 +++++++++++++++++++ .../plan/run_plan_pipeline.py | 10 +++ 4 files changed, 107 insertions(+) create mode 100644 worker_plan/worker_plan_internal/llm_util/usage_metrics.py diff --git a/worker_plan/worker_plan_api/filenames.py b/worker_plan/worker_plan_api/filenames.py index 82abbea89..3c9a8728f 100644 --- a/worker_plan/worker_plan_api/filenames.py +++ b/worker_plan/worker_plan_api/filenames.py @@ -126,3 +126,4 @@ class ExtraFilenameEnum(str, Enum): PIPELINE_STOP_REQUESTED_FLAG = "pipeline_stop_requested.txt" TRACK_ACTIVITY_JSONL = "track_activity.jsonl" ACTIVITY_OVERVIEW_JSON = "activity_overview.json" + USAGE_METRICS_JSONL = "usage_metrics.jsonl" diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index 65fa4d8dc..d702486fa 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -271,6 +271,25 @@ def _record_attempt_token_metrics( except Exception as exc: logger.debug("Failed to record token metrics for attempt: %s", exc) + # File-based metrics for local runs (no database required) + try: + from worker_plan_internal.llm_util.usage_metrics import record_usage_metric + from worker_plan_internal.llm_util.token_counter import extract_token_count + + token_count = extract_token_count(response) if response else None + record_usage_metric( + model=llm_model_name, + duration_seconds=duration, + success=success, + error_message=error_message, + input_tokens=token_count.input_tokens if token_count else None, + output_tokens=token_count.output_tokens if token_count else None, + thinking_tokens=token_count.thinking_tokens if token_count else None, + cost_usd=token_count.cost_usd if token_count else None, + ) + except Exception as exc: + logger.debug("Failed to record file-based usage metric: %s", exc) + def _check_stop_callback(self, last_attempt: LLMAttempt, start_time: float, attempt_index: int) -> None: """Checks the callback, if it exists, to see if execution should stop.""" if self.should_stop_callback is None: diff --git a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py new file mode 100644 index 000000000..2587ecd85 --- /dev/null +++ b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py @@ -0,0 +1,77 @@ +""" +File-based usage metrics for local runs. + +Records per-LLM-call metrics (model, tokens, duration, success/failure) to a JSONL file +in the run output directory. Works without a database — designed for local/offline runs. + +Usage: + from worker_plan_internal.llm_util.usage_metrics import set_usage_metrics_path, record_usage_metric + + # Set once at pipeline start + set_usage_metrics_path(run_id_dir / ExtraFilenameEnum.USAGE_METRICS_JSONL.value) + + # Called automatically by LLMExecutor._record_attempt_token_metrics() + record_usage_metric(model="gpt-4", duration=1.23, success=True, input_tokens=100, output_tokens=50) +""" +import json +import logging +from datetime import datetime +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +_usage_metrics_path: Optional[Path] = None + + +def set_usage_metrics_path(path: Optional[Path]) -> None: + """Set the JSONL file path for recording usage metrics.""" + global _usage_metrics_path + _usage_metrics_path = path + + +def get_usage_metrics_path() -> Optional[Path]: + """Get the current JSONL file path for recording usage metrics.""" + return _usage_metrics_path + + +def record_usage_metric( + model: str, + duration_seconds: float, + success: bool, + error_message: Optional[str] = None, + input_tokens: Optional[int] = None, + output_tokens: Optional[int] = None, + thinking_tokens: Optional[int] = None, + cost_usd: Optional[float] = None, +) -> None: + """Append a single usage metric record to the JSONL file. + + Best-effort: never raises exceptions to avoid blocking the LLM pipeline. + """ + path = _usage_metrics_path + if path is None: + return + + record = { + "timestamp": datetime.now().isoformat(), + "model": model, + "duration_seconds": round(duration_seconds, 3), + "success": success, + } + if error_message: + record["error"] = error_message + if input_tokens is not None: + record["input_tokens"] = input_tokens + if output_tokens is not None: + record["output_tokens"] = output_tokens + if thinking_tokens is not None: + record["thinking_tokens"] = thinking_tokens + if cost_usd is not None: + record["cost_usd"] = cost_usd + + try: + with open(path, "a", encoding="utf-8") as f: + f.write(json.dumps(record) + "\n") + except Exception as exc: + logger.debug("Failed to write usage metric: %s", exc) diff --git a/worker_plan/worker_plan_internal/plan/run_plan_pipeline.py b/worker_plan/worker_plan_internal/plan/run_plan_pipeline.py index fa5e8dbd7..e94568f1c 100644 --- a/worker_plan/worker_plan_internal/plan/run_plan_pipeline.py +++ b/worker_plan/worker_plan_internal/plan/run_plan_pipeline.py @@ -3947,6 +3947,7 @@ def get_progress_percentage(self) -> PipelineProgress: ExtraFilenameEnum.EXPECTED_FILENAMES1_JSON.value, ExtraFilenameEnum.LOG_TXT.value, ExtraFilenameEnum.PIPELINE_STOP_REQUESTED_FLAG.value, + ExtraFilenameEnum.USAGE_METRICS_JSONL.value, '.DS_Store', ] files = [f for f in files if f not in ignore_files] @@ -4028,6 +4029,12 @@ def run(self): logger.debug(f"Removing pre-existing stop flag file: {stop_flag_path!r}") stop_flag_path.unlink() + # Enable file-based usage metrics for this run. + from worker_plan_internal.llm_util.usage_metrics import set_usage_metrics_path + usage_metrics_path = self.run_id_dir / ExtraFilenameEnum.USAGE_METRICS_JSONL.value + set_usage_metrics_path(usage_metrics_path) + logger.info(f"Usage metrics will be written to {usage_metrics_path}") + # create a json file with the expected filenames. Save it to the run/run_id/expected_filenames1.json expected_filenames_path = self.run_id_dir / ExtraFilenameEnum.EXPECTED_FILENAMES1_JSON.value with open(expected_filenames_path, "w") as f: @@ -4042,6 +4049,9 @@ def run(self): workers=luigi_workers ) + # Clear the usage metrics path after the run. + set_usage_metrics_path(None) + # After the pipeline finishes (or fails), check for the stop flag. if self.has_stop_flag_file: logger.info("Pipeline was stopped intentionally via PipelineStopRequested exception.") From 7e1d5f802f8fe16e7b92da9117e1f17ffaca6bc1 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 12:42:18 +0100 Subject: [PATCH 02/12] Move usage_metrics and token_counter imports to top-level in llm_executor Fail hard on startup if imports are bad instead of silently swallowing errors inside a try block at runtime. Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_internal/llm_util/llm_executor.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index d702486fa..34d0573fe 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -39,6 +39,8 @@ from llama_index.core.llms.llm import LLM from llama_index.core.instrumentation.dispatcher import instrument_tags from worker_plan_internal.llm_factory import get_llm +from worker_plan_internal.llm_util.usage_metrics import record_usage_metric +from worker_plan_internal.llm_util.token_counter import extract_token_count logger = logging.getLogger(__name__) @@ -273,9 +275,6 @@ def _record_attempt_token_metrics( # File-based metrics for local runs (no database required) try: - from worker_plan_internal.llm_util.usage_metrics import record_usage_metric - from worker_plan_internal.llm_util.token_counter import extract_token_count - token_count = extract_token_count(response) if response else None record_usage_metric( model=llm_model_name, From 7344ed87ab251bf54bd70b6c637eac659c722542 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 12:47:08 +0100 Subject: [PATCH 03/12] Change usage metric recording failure log level from debug to warning Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_internal/llm_util/llm_executor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index 34d0573fe..e0aae1569 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -287,7 +287,7 @@ def _record_attempt_token_metrics( cost_usd=token_count.cost_usd if token_count else None, ) except Exception as exc: - logger.debug("Failed to record file-based usage metric: %s", exc) + logger.warning("Failed to record file-based usage metric: %s", exc) def _check_stop_callback(self, last_attempt: LLMAttempt, start_time: float, attempt_index: int) -> None: """Checks the callback, if it exists, to see if execution should stop.""" From 60c9bc5708aa81a4385381f8c3f138f5cf817353 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 12:57:33 +0100 Subject: [PATCH 04/12] Warn when usage metrics path is unset or write fails Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_internal/llm_util/usage_metrics.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py index 2587ecd85..0d46b6710 100644 --- a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py +++ b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py @@ -51,6 +51,7 @@ def record_usage_metric( """ path = _usage_metrics_path if path is None: + logger.warning("record_usage_metric called but no usage metrics path is set") return record = { @@ -74,4 +75,4 @@ def record_usage_metric( with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record) + "\n") except Exception as exc: - logger.debug("Failed to write usage metric: %s", exc) + logger.warning("Failed to write usage metric: %s", exc) From b1d5e3d9ea98ae6d6b012e1ddfb97495bc519f00 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 13:03:33 +0100 Subject: [PATCH 05/12] Document set_usage_metrics_path(None) teardown in usage example Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_internal/llm_util/usage_metrics.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py index 0d46b6710..52bf08abd 100644 --- a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py +++ b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py @@ -12,6 +12,9 @@ # Called automatically by LLMExecutor._record_attempt_token_metrics() record_usage_metric(model="gpt-4", duration=1.23, success=True, input_tokens=100, output_tokens=50) + + # Clear after pipeline completes to avoid stale state + set_usage_metrics_path(None) """ import json import logging From f7a1dd8362684ad98e0b131096c6ede974f99128 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 13:05:39 +0100 Subject: [PATCH 06/12] Remove redundant try/except around usage metric recording record_usage_metric and extract_token_count handle errors internally, so the outer try block was unnecessary. Co-Authored-By: Claude Opus 4.6 --- .../llm_util/llm_executor.py | 25 ++++++++----------- 1 file changed, 11 insertions(+), 14 deletions(-) diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index e0aae1569..ccf6c92ad 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -274,20 +274,17 @@ def _record_attempt_token_metrics( logger.debug("Failed to record token metrics for attempt: %s", exc) # File-based metrics for local runs (no database required) - try: - token_count = extract_token_count(response) if response else None - record_usage_metric( - model=llm_model_name, - duration_seconds=duration, - success=success, - error_message=error_message, - input_tokens=token_count.input_tokens if token_count else None, - output_tokens=token_count.output_tokens if token_count else None, - thinking_tokens=token_count.thinking_tokens if token_count else None, - cost_usd=token_count.cost_usd if token_count else None, - ) - except Exception as exc: - logger.warning("Failed to record file-based usage metric: %s", exc) + token_count = extract_token_count(response) if response else None + record_usage_metric( + model=llm_model_name, + duration_seconds=duration, + success=success, + error_message=error_message, + input_tokens=token_count.input_tokens if token_count else None, + output_tokens=token_count.output_tokens if token_count else None, + thinking_tokens=token_count.thinking_tokens if token_count else None, + cost_usd=token_count.cost_usd if token_count else None, + ) def _check_stop_callback(self, last_attempt: LLMAttempt, start_time: float, attempt_index: int) -> None: """Checks the callback, if it exists, to see if execution should stop.""" From 7ccfade1e3921ba295579f8fa425e884ef1173fa Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 13:40:20 +0100 Subject: [PATCH 07/12] Allow resume for legacy plans without pipeline_version in parameters Plans created before pipeline_version was stamped into parameters were incorrectly rejected by the frontend and MCP resume checks. The worker-side check against the actual snapshot metadata file is the real safety gate, so allow None through at the API layer. Co-Authored-By: Claude Opus 4.6 --- frontend_multi_user/src/app.py | 4 +++- mcp_cloud/db_queries.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/frontend_multi_user/src/app.py b/frontend_multi_user/src/app.py index b00431fa6..b9db3ebe9 100644 --- a/frontend_multi_user/src/app.py +++ b/frontend_multi_user/src/app.py @@ -3405,9 +3405,11 @@ def plan_resume(): return redirect(url_for('plan', id=run_id)) # Reject resume if the snapshot was created by a different pipeline version. + # Plans created before pipeline_version was stamped have no stored version; + # allow those through — the worker-side metadata check is the real safety gate. stored_params = task.parameters if isinstance(task.parameters, dict) else {} stored_version = stored_params.get("pipeline_version") - if stored_version != PIPELINE_VERSION: + if stored_version is not None and stored_version != PIPELINE_VERSION: return redirect(url_for( 'plan', id=run_id, resume_error="version_mismatch", diff --git a/mcp_cloud/db_queries.py b/mcp_cloud/db_queries.py index a87142416..e57db7fe9 100644 --- a/mcp_cloud/db_queries.py +++ b/mcp_cloud/db_queries.py @@ -268,9 +268,11 @@ def _resume_plan_sync(plan_id: str, model_profile: str) -> Optional[dict[str, An } # Reject resume if the snapshot was created by a different pipeline version. + # Plans created before pipeline_version was stamped have no stored version; + # allow those through — the worker-side metadata check is the real safety gate. stored_params = plan.parameters if isinstance(plan.parameters, dict) else {} stored_version = stored_params.get("pipeline_version") - if stored_version != PIPELINE_VERSION: + if stored_version is not None and stored_version != PIPELINE_VERSION: return { "error": { "code": "PIPELINE_VERSION_MISMATCH", From 253bab4bdca97f644472fd0b1e61ddffd3d40bb9 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 13:52:10 +0100 Subject: [PATCH 08/12] Elaborate on worker-side version check in resume comments Point to worker_plan_database/app.py and the actual snapshot metadata file (001-3-planexe_metadata.json) so readers can find the real check. Co-Authored-By: Claude Opus 4.6 --- frontend_multi_user/src/app.py | 4 +++- mcp_cloud/db_queries.py | 4 +++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/frontend_multi_user/src/app.py b/frontend_multi_user/src/app.py index b9db3ebe9..77f910d81 100644 --- a/frontend_multi_user/src/app.py +++ b/frontend_multi_user/src/app.py @@ -3406,7 +3406,9 @@ def plan_resume(): # Reject resume if the snapshot was created by a different pipeline version. # Plans created before pipeline_version was stamped have no stored version; - # allow those through — the worker-side metadata check is the real safety gate. + # allow those through — the worker-side check in worker_plan_database/app.py + # reads pipeline_version from the actual snapshot metadata file + # (001-3-planexe_metadata.json) and rejects incompatible versions there. stored_params = task.parameters if isinstance(task.parameters, dict) else {} stored_version = stored_params.get("pipeline_version") if stored_version is not None and stored_version != PIPELINE_VERSION: diff --git a/mcp_cloud/db_queries.py b/mcp_cloud/db_queries.py index e57db7fe9..6f03c3cb7 100644 --- a/mcp_cloud/db_queries.py +++ b/mcp_cloud/db_queries.py @@ -269,7 +269,9 @@ def _resume_plan_sync(plan_id: str, model_profile: str) -> Optional[dict[str, An # Reject resume if the snapshot was created by a different pipeline version. # Plans created before pipeline_version was stamped have no stored version; - # allow those through — the worker-side metadata check is the real safety gate. + # allow those through — the worker-side check in worker_plan_database/app.py + # reads pipeline_version from the actual snapshot metadata file + # (001-3-planexe_metadata.json) and rejects incompatible versions there. stored_params = plan.parameters if isinstance(plan.parameters, dict) else {} stored_version = stored_params.get("pipeline_version") if stored_version is not None and stored_version != PIPELINE_VERSION: From 5eb928c496457b060078a91c6da1f8aa636eda15 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 13:58:24 +0100 Subject: [PATCH 09/12] Add upstream_provider and upstream_model to usage metrics The detailed provider/model info (e.g. "Google AI Studio:google/gemini-2.0-flash-001") was already extracted by token_counter but not recorded. Now included in usage_metrics.jsonl when available. Also move success field before model for easier error skimming. Co-Authored-By: Claude Opus 4.6 --- worker_plan/worker_plan_internal/llm_util/llm_executor.py | 2 ++ .../worker_plan_internal/llm_util/usage_metrics.py | 8 +++++++- 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index ccf6c92ad..8596aa86b 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -284,6 +284,8 @@ def _record_attempt_token_metrics( output_tokens=token_count.output_tokens if token_count else None, thinking_tokens=token_count.thinking_tokens if token_count else None, cost_usd=token_count.cost_usd if token_count else None, + upstream_provider=token_count.upstream_provider if token_count else None, + upstream_model=token_count.upstream_model if token_count else None, ) def _check_stop_callback(self, last_attempt: LLMAttempt, start_time: float, attempt_index: int) -> None: diff --git a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py index 52bf08abd..553c0b576 100644 --- a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py +++ b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py @@ -47,6 +47,8 @@ def record_usage_metric( output_tokens: Optional[int] = None, thinking_tokens: Optional[int] = None, cost_usd: Optional[float] = None, + upstream_provider: Optional[str] = None, + upstream_model: Optional[str] = None, ) -> None: """Append a single usage metric record to the JSONL file. @@ -59,9 +61,9 @@ def record_usage_metric( record = { "timestamp": datetime.now().isoformat(), + "success": success, "model": model, "duration_seconds": round(duration_seconds, 3), - "success": success, } if error_message: record["error"] = error_message @@ -73,6 +75,10 @@ def record_usage_metric( record["thinking_tokens"] = thinking_tokens if cost_usd is not None: record["cost_usd"] = cost_usd + if upstream_provider is not None: + record["upstream_provider"] = upstream_provider + if upstream_model is not None: + record["upstream_model"] = upstream_model try: with open(path, "a", encoding="utf-8") as f: From 39f451aff0508f5d348a88c7fac98b719bea3672 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 14:38:08 +0100 Subject: [PATCH 10/12] Record usage metrics from TrackActivity instrumentation events Successful LLM calls are now recorded by TrackActivity which has access to the real ChatResponse with full token counts, cost, and provider:model info. LLMExecutor only records failures since instrumentation end events are not emitted when the call fails. Skip events without token usage or cost to avoid "unknown" rows. Remove redundant upstream_provider and upstream_model fields since model already contains "provider:model". Co-Authored-By: Claude Opus 4.6 --- .../llm_util/llm_executor.py | 28 +++++++++---------- .../llm_util/track_activity.py | 18 ++++++++++++ .../llm_util/usage_metrics.py | 6 ---- 3 files changed, 31 insertions(+), 21 deletions(-) diff --git a/worker_plan/worker_plan_internal/llm_util/llm_executor.py b/worker_plan/worker_plan_internal/llm_util/llm_executor.py index 8596aa86b..a76601c37 100644 --- a/worker_plan/worker_plan_internal/llm_util/llm_executor.py +++ b/worker_plan/worker_plan_internal/llm_util/llm_executor.py @@ -40,7 +40,6 @@ from llama_index.core.instrumentation.dispatcher import instrument_tags from worker_plan_internal.llm_factory import get_llm from worker_plan_internal.llm_util.usage_metrics import record_usage_metric -from worker_plan_internal.llm_util.token_counter import extract_token_count logger = logging.getLogger(__name__) @@ -273,20 +272,19 @@ def _record_attempt_token_metrics( except Exception as exc: logger.debug("Failed to record token metrics for attempt: %s", exc) - # File-based metrics for local runs (no database required) - token_count = extract_token_count(response) if response else None - record_usage_metric( - model=llm_model_name, - duration_seconds=duration, - success=success, - error_message=error_message, - input_tokens=token_count.input_tokens if token_count else None, - output_tokens=token_count.output_tokens if token_count else None, - thinking_tokens=token_count.thinking_tokens if token_count else None, - cost_usd=token_count.cost_usd if token_count else None, - upstream_provider=token_count.upstream_provider if token_count else None, - upstream_model=token_count.upstream_model if token_count else None, - ) + # File-based usage metrics for local runs (no database required). + # Successful calls are recorded by TrackActivity via llama_index + # instrumentation, which has access to the real ChatResponse with + # full token counts and upstream provider/model info. + # Here we only record failures, since instrumentation end events + # are not emitted when the LLM call fails. + if not success: + record_usage_metric( + model=llm_model_name, + duration_seconds=duration, + success=False, + error_message=error_message, + ) def _check_stop_callback(self, last_attempt: LLMAttempt, start_time: float, attempt_index: int) -> None: """Checks the callback, if it exists, to see if execution should stop.""" diff --git a/worker_plan/worker_plan_internal/llm_util/track_activity.py b/worker_plan/worker_plan_internal/llm_util/track_activity.py index 29b0d03e7..f583f5ffb 100644 --- a/worker_plan/worker_plan_internal/llm_util/track_activity.py +++ b/worker_plan/worker_plan_internal/llm_util/track_activity.py @@ -299,6 +299,23 @@ def _record_token_metrics_row(self, event_data: dict, duration_seconds: Optional except Exception: logger.debug("Failed to persist token metrics from TrackActivity", exc_info=True) + def _record_file_usage_metric(self, event_data: dict, token_usage: Optional[dict], duration_seconds: Optional[float]) -> None: + """Write a usage metric row to usage_metrics.jsonl via the shared module.""" + cost = self._extract_cost(event_data) + if not token_usage and cost == 0.0: + return + from worker_plan_internal.llm_util.usage_metrics import record_usage_metric + model_name = self._extract_model_name(event_data) + record_usage_metric( + model=model_name or "unknown", + duration_seconds=duration_seconds or 0.0, + success=True, + input_tokens=int(token_usage["input_tokens"]) if token_usage and token_usage.get("input_tokens") is not None else None, + output_tokens=int(token_usage["output_tokens"]) if token_usage and token_usage.get("output_tokens") is not None else None, + thinking_tokens=int(token_usage["thinking_tokens"]) if token_usage and token_usage.get("thinking_tokens") is not None else None, + cost_usd=cost or None, + ) + @staticmethod def _split_provider_and_model(model_name: str) -> tuple[Optional[str], Optional[str]]: if not model_name: @@ -367,6 +384,7 @@ def handle(self, event: Any) -> None: event_record["token_usage"] = token_usage self._update_activity_overview(filtered_event_data) self._record_token_metrics_row(filtered_event_data, duration_seconds=duration_seconds) + self._record_file_usage_metric(filtered_event_data, token_usage, duration_seconds) elif isinstance(event, (LLMChatStartEvent, LLMCompletionStartEvent, LLMStructuredPredictStartEvent)): match_key = self._event_match_key(filtered_event_data) start_ts = self._parse_event_timestamp(filtered_event_data) diff --git a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py index 553c0b576..9e83d44a8 100644 --- a/worker_plan/worker_plan_internal/llm_util/usage_metrics.py +++ b/worker_plan/worker_plan_internal/llm_util/usage_metrics.py @@ -47,8 +47,6 @@ def record_usage_metric( output_tokens: Optional[int] = None, thinking_tokens: Optional[int] = None, cost_usd: Optional[float] = None, - upstream_provider: Optional[str] = None, - upstream_model: Optional[str] = None, ) -> None: """Append a single usage metric record to the JSONL file. @@ -75,10 +73,6 @@ def record_usage_metric( record["thinking_tokens"] = thinking_tokens if cost_usd is not None: record["cost_usd"] = cost_usd - if upstream_provider is not None: - record["upstream_provider"] = upstream_provider - if upstream_model is not None: - record["upstream_model"] = upstream_model try: with open(path, "a", encoding="utf-8") as f: From 9a5fbb99451cbe4c2774c3be67971bcf5ed4e776 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 14:42:47 +0100 Subject: [PATCH 11/12] Update proposals 110 and 111 to reflect implementation status Mark proposal 110 as implemented with PR #219 details: JSONL format, instrumentation-based recording, resolved open questions. Mark 110 as complete in the promising directions roadmap. Co-Authored-By: Claude Opus 4.6 --- .../proposals/110-usage-metrics-local-runs.md | 58 +++++++++++++++---- docs/proposals/111-promising-directions.md | 4 +- 2 files changed, 48 insertions(+), 14 deletions(-) diff --git a/docs/proposals/110-usage-metrics-local-runs.md b/docs/proposals/110-usage-metrics-local-runs.md index e9e9eb09b..078fd4951 100644 --- a/docs/proposals/110-usage-metrics-local-runs.md +++ b/docs/proposals/110-usage-metrics-local-runs.md @@ -2,7 +2,7 @@ **Author:** Bubba (VoynichLabs) **Date:** 2026-03-08 -**Status:** Proposal +**Status:** Implemented (PR #219) --- @@ -165,18 +165,52 @@ None of these are present in a bare CLI run. The file-based approach requires on --- -## Open Questions +## Implementation Notes (PR #219) -1. **Ollama field mapping**: `prompt_eval_count` / `eval_count` vs `prompt_tokens` / - `completion_tokens` — should `extract_token_count()` add an Ollama-specific branch, or - should the Ollama LLM adapter normalize field names before returning? +The final implementation diverges from Option A above in several ways: -2. **Thread safety**: Luigi runs tasks in parallel workers. The appender needs an - `fcntl.flock()` or atomic rename approach to avoid corruption under concurrent writes. +### File format: JSONL, not JSON -3. **Thinking tokens for local models**: Qwen 3 thinking tokens are stripped before the - response reaches llama_index (LM Studio does not expose them separately). The - `thinking_tokens` field will be `null` for local runs — acceptable for now. +`usage_metrics.jsonl` uses one JSON object per line (append-only). This avoids the need for +atomic rename or in-memory accumulation — each LLM call appends a single line. Thread-safe +by nature since each write is a short append to a file handle. -4. **Retention**: Should `usage_metrics.json` be overwritten on resume (Luigi skip-completed - behavior) or appended? Appending is safer — a resumed run adds only the new calls. +### Recording source: llama_index instrumentation, not LLMExecutor + +Successful calls are recorded by `TrackActivity` (the llama_index `BaseEventHandler`) which +receives the actual `ChatResponse` with full token counts, cost, and `provider:model` info. +`LLMExecutor._record_attempt_token_metrics()` only records **failures**, since instrumentation +end events are not emitted when the LLM call fails. + +This was necessary because `execute_function(llm)` returns the processed result (a Pydantic +model or string), not the raw `ChatResponse`. The instrumentation layer is the only place +with access to the real response. + +### Model field includes provider + +The `model` field contains the full `provider:model` string (e.g. +`Google AI Studio:google/gemini-2.0-flash-001`), matching `activity_overview.json`. + +### Example output + +```json +{"timestamp": "2026-03-10T13:36:48.250446", "success": true, "model": "Google AI Studio:google/gemini-2.0-flash-001", "duration_seconds": 4.879, "input_tokens": 5316, "output_tokens": 643, "cost_usd": 0.0007888} +{"timestamp": "2026-03-10T13:36:53.554864", "success": true, "model": "Google:google/gemini-2.0-flash-001", "duration_seconds": 5.237, "input_tokens": 8877, "output_tokens": 562, "cost_usd": 0.0011125} +``` + +### Key files + +| File | Role | +|------|------| +| `worker_plan/worker_plan_internal/llm_util/usage_metrics.py` | Core module: `set_usage_metrics_path()`, `record_usage_metric()` | +| `worker_plan/worker_plan_internal/llm_util/track_activity.py` | Records successful calls via `_record_file_usage_metric()` | +| `worker_plan/worker_plan_internal/llm_util/llm_executor.py` | Records failed calls only | +| `worker_plan/worker_plan_internal/plan/run_plan_pipeline.py` | Sets/clears metrics path around pipeline execution | +| `worker_plan/worker_plan_api/filenames.py` | `USAGE_METRICS_JSONL` constant | + +### Resolved open questions + +1. **Ollama field mapping**: Handled by `extract_token_count()` and `TrackActivity._extract_token_usage()` which already support multiple field name variations. +2. **Thread safety**: JSONL append-per-line is safe for concurrent Luigi workers. +3. **Thinking tokens**: Recorded when available (e.g. from OpenRouter reasoning models). `null` for providers that don't expose them. +4. **Retention on resume**: Appended. A resumed run adds only the new calls alongside the restored snapshot's existing metrics. diff --git a/docs/proposals/111-promising-directions.md b/docs/proposals/111-promising-directions.md index fd8fbf2d7..53b25f9b3 100644 --- a/docs/proposals/111-promising-directions.md +++ b/docs/proposals/111-promising-directions.md @@ -34,7 +34,7 @@ Agents need to discover PlanExe, understand its tools, and consume outputs progr |---|----------|-------------| | **86** | Agent-Optimized Pipeline | Removes the 5 key friction points for autonomous agent use: human approval gate, no agent prompt examples, poll intervals tuned for humans, no machine-readable output, no autonomous agent setup docs | | **62** | Agent-First Frontend Discoverability | `llms.txt`, `/.well-known/mcp.json`, agent-readable README — standard discovery protocols so agents find PlanExe without human guidance | -| **110** | Usage Metrics for Local Runs | Agents need cost accounting for budget-constrained workflows. Answers "how much did this run cost?" | +| **110** | Usage Metrics for Local Runs | ✅ **Implemented (PR #219)**. Agents need cost accounting for budget-constrained workflows. `usage_metrics.jsonl` answers "how much did this run cost?" with per-call granularity (model, tokens, cost, duration). Complements `activity_overview.json` aggregated totals | Key friction points from #86 that block autonomous agent use: - **F1**: Human approval step before `plan_create` — autonomous agents can't proceed @@ -94,7 +94,7 @@ Phase 1: Reliable foundation (now) ├─ #87 Plan resume ├─ #109 Retry improvements ├─ #102 Error-feedback retries - ├─ #110 Usage metrics + ├─ #110 Usage metrics ✅ └─ #58 Prompt boost Phase 2: Agent-native interface (next) From b9fe5d2b5549527ae42a14e0528f6421f9ef4222 Mon Sep 17 00:00:00 2001 From: Simon Strandgaard Date: Tue, 10 Mar 2026 15:17:25 +0100 Subject: [PATCH 12/12] Protect pipeline from non-fatal heartbeat upsert failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A corrupted psycopg2 connection during WorkerItem.upsert_heartbeat() was propagating up and killing Luigi tasks. The heartbeat is just a liveness signal — wrap it in try/except with a session rollback so the pipeline can continue. Co-Authored-By: Claude Opus 4.6 --- worker_plan_database/app.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/worker_plan_database/app.py b/worker_plan_database/app.py index 85620ec81..71e35fc48 100644 --- a/worker_plan_database/app.py +++ b/worker_plan_database/app.py @@ -442,7 +442,14 @@ def _handle_task_completion(self, parameters: HandleTaskCompletionParameters) -> """ logger.debug(f"ServerExecutePipeline._handle_task_completion") - WorkerItem.upsert_heartbeat(worker_id=WORKER_ID, current_task_id=self.task_id) + try: + WorkerItem.upsert_heartbeat(worker_id=WORKER_ID, current_task_id=self.task_id) + except Exception as exc: + logger.warning("Heartbeat upsert failed (non-fatal): %s", exc) + try: + db.session.rollback() + except Exception: + pass # Lookup the taskitem in the database by self.task_id task = db.session.get(PlanItem, self.task_id)