mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-31 19:16:29 +00:00
Merge pull request #73544 from afourniernv/fix/relay-desktop-outbox-export
fix(observability): export shared metrics after task completion
This commit is contained in:
Jeffrey Quesnelle
GitHub
commit
38574a7397
4 changed files with 234 additions and 13 deletions
|
|
@ -351,7 +351,17 @@ class _Runtime:
|
|||
with session.lock:
|
||||
if session.closing:
|
||||
return
|
||||
self._finish_task(session, task_id, event)
|
||||
finished = self._finish_task(session, task_id, event)
|
||||
if finished:
|
||||
try:
|
||||
self.relay.subscribers.flush()
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Hermes shared-metrics task flush failed",
|
||||
exc_info=True,
|
||||
)
|
||||
else:
|
||||
self._export()
|
||||
|
||||
def close_session(self, event: dict[str, Any]) -> None:
|
||||
session = self._session(event)
|
||||
|
|
@ -380,7 +390,8 @@ class _Runtime:
|
|||
self.relay.subscribers.flush()
|
||||
except Exception as exc:
|
||||
failures.append(f"subscriber flush failed: {exc}")
|
||||
self._export()
|
||||
else:
|
||||
self._export()
|
||||
with self._sessions_lock:
|
||||
if self._sessions.get(session.session_id) is session:
|
||||
self._sessions.pop(session.session_id, None)
|
||||
|
|
@ -399,8 +410,15 @@ class _Runtime:
|
|||
self._safe(self.close_session, {"session_id": session_id})
|
||||
if not self._registered:
|
||||
return
|
||||
self._safe(self.relay.subscribers.flush)
|
||||
self._export()
|
||||
try:
|
||||
self.relay.subscribers.flush()
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Hermes shared-metrics shutdown flush failed",
|
||||
exc_info=True,
|
||||
)
|
||||
else:
|
||||
self._export()
|
||||
self._safe(self.relay.subscribers.deregister, self._subscriber_name)
|
||||
self.host.release_managed_execution(self._subscriber_name)
|
||||
self._registered = False
|
||||
|
|
@ -560,10 +578,10 @@ class _Runtime:
|
|||
session: _MetricsSession,
|
||||
task_id: str,
|
||||
event: dict[str, Any],
|
||||
) -> None:
|
||||
) -> bool:
|
||||
task = session.tasks.get(task_id)
|
||||
if task is None:
|
||||
return
|
||||
return False
|
||||
self._end_pending_model_calls(session, {**event, "task_id": task_id})
|
||||
fields = task_terminal_fields(
|
||||
{**task.start_fields, **event},
|
||||
|
|
@ -592,9 +610,10 @@ class _Runtime:
|
|||
turn_key = (session.session_id, turn_id)
|
||||
if self._turn_sessions.get(turn_key) is session:
|
||||
self._turn_sessions.pop(turn_key, None)
|
||||
return True
|
||||
|
||||
def _export(self) -> None:
|
||||
self._safe(self.subscriber.store.create_and_export_package)
|
||||
self._safe(self.subscriber.store.create_and_export_package_if_due)
|
||||
|
||||
def _event_metadata(self) -> dict[str, str]:
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -114,6 +114,14 @@ class SharedMetricsStore:
|
|||
for _ in range(pending_periods):
|
||||
if self._create_package() is None:
|
||||
break
|
||||
return self._export_and_prune()
|
||||
|
||||
def create_and_export_package_if_due(self) -> list[Path]:
|
||||
"""Create pending packages at most once per UTC day, then export them."""
|
||||
self._create_pending_packages_if_due()
|
||||
return self._export_and_prune()
|
||||
|
||||
def _export_and_prune(self) -> list[Path]:
|
||||
exported = self._export_pending_packages()
|
||||
try:
|
||||
self._prune_expired_history()
|
||||
|
|
@ -281,6 +289,26 @@ class SharedMetricsStore:
|
|||
).fetchone()
|
||||
return int(row["period_count"]) if row is not None else 0
|
||||
|
||||
def _create_pending_packages_if_due(self) -> None:
|
||||
now = _utc_now()
|
||||
with self._connection() as connection:
|
||||
with write_txn(connection):
|
||||
# Gate on the committed package, not its file write, so a failed
|
||||
# outbox export can be retried without packaging deltas twice.
|
||||
package_created_today = connection.execute(
|
||||
"""
|
||||
SELECT 1
|
||||
FROM package_outbox
|
||||
WHERE substr(created_at, 1, 10) >= ?
|
||||
LIMIT 1
|
||||
""",
|
||||
(now.date().isoformat(),),
|
||||
).fetchone()
|
||||
if package_created_today is not None:
|
||||
return
|
||||
while self._create_package_in_transaction(connection, now) is not None:
|
||||
pass
|
||||
|
||||
def _create_package(self) -> dict[str, Any] | None:
|
||||
now = _utc_now()
|
||||
with self._connection() as connection:
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ from types import SimpleNamespace
|
|||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from hermes_cli.observability import shared_metrics as shared_metrics_module
|
||||
from hermes_cli.observability.shared_metrics import SharedMetricsStore
|
||||
from hermes_cli.observability.shared_metrics_contract import (
|
||||
COUNT_BUCKETS,
|
||||
|
|
@ -142,6 +143,43 @@ def test_model_call_counter_survives_restart_and_exports_only_new_deltas(tmp_pat
|
|||
assert restarted.counter_snapshot()[0]["packaged_value"] == 3
|
||||
|
||||
|
||||
def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas(
|
||||
tmp_path, monkeypatch
|
||||
):
|
||||
current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc)
|
||||
monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: current_time)
|
||||
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
|
||||
|
||||
store.record_model_call(_dimensions(), "test-version")
|
||||
assert len(store.create_and_export_package_if_due()) == 1
|
||||
|
||||
current_time = datetime(2026, 7, 28, 18, tzinfo=timezone.utc)
|
||||
store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
|
||||
store.record_model_call(_dimensions(), "test-version")
|
||||
assert store.create_and_export_package_if_due() == []
|
||||
assert len(list((tmp_path / "outbox").glob("*.json"))) == 1
|
||||
assert store.counter_snapshot()[0] == {
|
||||
"period_start": "2026-07-28",
|
||||
"metric_name": "hermes.model_call.count",
|
||||
"hermes_version": "test-version",
|
||||
"dimensions": _dimensions(),
|
||||
"value": 2,
|
||||
"packaged_value": 1,
|
||||
}
|
||||
|
||||
current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
|
||||
store.record_model_call(_dimensions(), "test-version")
|
||||
assert len(store.create_and_export_package_if_due()) == 2
|
||||
assert len(list((tmp_path / "outbox").glob("*.json"))) == 3
|
||||
assert all(
|
||||
row["value"] == row["packaged_value"] for row in store.counter_snapshot()
|
||||
)
|
||||
|
||||
store.record_model_call(_dimensions(), "test-version")
|
||||
assert store.create_and_export_package_if_due() == []
|
||||
assert len(list((tmp_path / "outbox").glob("*.json"))) == 3
|
||||
|
||||
|
||||
def test_package_schema_matches_the_model_call_contract():
|
||||
properties = _package_dimension_schema()["properties"]
|
||||
|
||||
|
|
@ -758,6 +796,32 @@ def test_concurrent_package_builders_commit_one_delta(tmp_path):
|
|||
assert store.counter_snapshot()[0]["packaged_value"] == 1
|
||||
|
||||
|
||||
def test_concurrent_due_exports_create_one_daily_package(tmp_path):
|
||||
database_path = tmp_path / "metrics.sqlite3"
|
||||
outbox_directory = tmp_path / "outbox"
|
||||
store = SharedMetricsStore(database_path, outbox_directory)
|
||||
store.record_model_call(_dimensions(), "test-version")
|
||||
ready = threading.Barrier(8)
|
||||
|
||||
def export() -> None:
|
||||
worker_store = SharedMetricsStore(database_path, outbox_directory)
|
||||
ready.wait(timeout=5)
|
||||
worker_store.create_and_export_package_if_due()
|
||||
|
||||
with ThreadPoolExecutor(max_workers=8) as executor:
|
||||
futures = [executor.submit(export) for _ in range(8)]
|
||||
for future in futures:
|
||||
future.result()
|
||||
|
||||
with sqlite3.connect(database_path) as connection:
|
||||
[outbox_count] = connection.execute(
|
||||
"SELECT COUNT(*) FROM package_outbox"
|
||||
).fetchone()
|
||||
assert outbox_count == 1
|
||||
assert len(list(outbox_directory.glob("*.json"))) == 1
|
||||
assert store.counter_snapshot()[0]["packaged_value"] == 1
|
||||
|
||||
|
||||
def test_concurrent_model_call_updates_are_transactional(tmp_path):
|
||||
database_path = tmp_path / "metrics.sqlite3"
|
||||
outbox_directory = tmp_path / "outbox"
|
||||
|
|
|
|||
|
|
@ -5,7 +5,9 @@ from __future__ import annotations
|
|||
import contextvars
|
||||
import asyncio
|
||||
import json
|
||||
import sqlite3
|
||||
import threading
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from typing import Any
|
||||
|
|
@ -321,6 +323,7 @@ def test_direct_runtime_records_without_enabling_a_plugin(direct_runtime, tmp_pa
|
|||
def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot(
|
||||
real_binding_runtime,
|
||||
tmp_path,
|
||||
monkeypatch,
|
||||
):
|
||||
assert real_binding_runtime._native is not None
|
||||
prompt_canary = "real-relay-sensitive-prompt"
|
||||
|
|
@ -416,6 +419,12 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot(
|
|||
|
||||
root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics"
|
||||
store = SharedMetricsStore(root / "metrics.sqlite3", root / "outbox")
|
||||
tomorrow = datetime.now(timezone.utc) + timedelta(days=1)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.observability.shared_metrics._utc_now",
|
||||
lambda: tomorrow,
|
||||
)
|
||||
assert len(store.create_and_export_package_if_due()) == 1
|
||||
snapshot = store.counter_snapshot()
|
||||
by_metric: dict[str, list[dict[str, Any]]] = {}
|
||||
for counter in snapshot:
|
||||
|
|
@ -451,7 +460,7 @@ def test_real_binding_drives_lifecycle_aggregation_export_and_snapshot(
|
|||
}
|
||||
package_values: dict[tuple[str, tuple[tuple[str, str], ...]], int] = {}
|
||||
packages = sorted((root / "outbox").glob("*.json"))
|
||||
assert len(packages) == 3
|
||||
assert len(packages) == 2
|
||||
package_payloads = [
|
||||
json.loads(package.read_text(encoding="utf-8")) for package in packages
|
||||
]
|
||||
|
|
@ -2279,32 +2288,133 @@ def test_session_finalize_closes_a_pending_task_as_system_aborted(direct_runtime
|
|||
}
|
||||
|
||||
|
||||
def test_sequential_tasks_in_one_session_aggregate_once_each(direct_runtime, tmp_path):
|
||||
def test_desktop_task_completion_exports_once_per_utc_day(
|
||||
direct_runtime, tmp_path, monkeypatch
|
||||
):
|
||||
current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.observability.shared_metrics._utc_now",
|
||||
lambda: current_time,
|
||||
)
|
||||
for task_id in ("t1", "t2"):
|
||||
lifecycle.invoke_hook(
|
||||
"pre_llm_call",
|
||||
session_id="s1",
|
||||
task_id=task_id,
|
||||
platform="cli",
|
||||
platform="desktop",
|
||||
)
|
||||
lifecycle.invoke_hook(
|
||||
"on_session_end",
|
||||
session_id="s1",
|
||||
task_id=task_id,
|
||||
platform="cli",
|
||||
platform="desktop",
|
||||
completed=True,
|
||||
failed=False,
|
||||
interrupted=False,
|
||||
turn_exit_reason="text_response(stop)",
|
||||
)
|
||||
lifecycle.finalize_session(session_id="s1")
|
||||
|
||||
outbox = tmp_path / "hermes-home" / "telemetry" / "shared_metrics" / "outbox"
|
||||
[package_path] = list(outbox.glob("*.json"))
|
||||
[first_package_path] = list(outbox.glob("*.json"))
|
||||
first_package = json.loads(first_package_path.read_text(encoding="utf-8"))
|
||||
first_metrics = {metric["name"]: metric for metric in first_package["metrics"]}
|
||||
assert first_metrics["hermes.task_run.started"]["value"] == 1
|
||||
assert first_metrics["hermes.task_run.started"]["dimensions"] == {
|
||||
"entrypoint": "interactive",
|
||||
"execution_surface": "desktop",
|
||||
}
|
||||
assert first_metrics["hermes.task_run.finished"]["value"] == 1
|
||||
|
||||
lifecycle.finalize_session(session_id="s1")
|
||||
assert list(outbox.glob("*.json")) == [first_package_path]
|
||||
|
||||
current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
|
||||
lifecycle.invoke_hook(
|
||||
"pre_llm_call",
|
||||
session_id="s1",
|
||||
task_id="t3",
|
||||
platform="desktop",
|
||||
)
|
||||
lifecycle.invoke_hook(
|
||||
"on_session_end",
|
||||
session_id="s1",
|
||||
task_id="t3",
|
||||
platform="desktop",
|
||||
completed=True,
|
||||
failed=False,
|
||||
interrupted=False,
|
||||
turn_exit_reason="text_response(stop)",
|
||||
)
|
||||
|
||||
packages = [
|
||||
json.loads(package_path.read_text(encoding="utf-8"))
|
||||
for package_path in outbox.glob("*.json")
|
||||
]
|
||||
totals: dict[str, int] = {}
|
||||
for package in packages:
|
||||
for metric in package["metrics"]:
|
||||
totals[metric["name"]] = totals.get(metric["name"], 0) + metric["value"]
|
||||
assert totals["hermes.task_run.started"] == 3
|
||||
assert totals["hermes.task_run.finished"] == 3
|
||||
|
||||
|
||||
def test_failed_flush_keeps_daily_export_open_for_later_task(
|
||||
direct_runtime, tmp_path, monkeypatch, caplog
|
||||
):
|
||||
current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc)
|
||||
monkeypatch.setattr(
|
||||
"hermes_cli.observability.shared_metrics._utc_now",
|
||||
lambda: current_time,
|
||||
)
|
||||
original_flush = direct_runtime.subscribers.flush
|
||||
flush_attempts = 0
|
||||
|
||||
def fail_first_flush() -> None:
|
||||
nonlocal flush_attempts
|
||||
flush_attempts += 1
|
||||
if flush_attempts == 1:
|
||||
raise RuntimeError("simulated flush failure")
|
||||
original_flush()
|
||||
|
||||
direct_runtime.subscribers.flush = fail_first_flush
|
||||
|
||||
def finish_desktop_task(task_id: str) -> None:
|
||||
lifecycle.invoke_hook(
|
||||
"pre_llm_call",
|
||||
session_id="s1",
|
||||
task_id=task_id,
|
||||
platform="desktop",
|
||||
)
|
||||
lifecycle.invoke_hook(
|
||||
"on_session_end",
|
||||
session_id="s1",
|
||||
task_id=task_id,
|
||||
platform="desktop",
|
||||
completed=True,
|
||||
failed=False,
|
||||
interrupted=False,
|
||||
turn_exit_reason="text_response(stop)",
|
||||
)
|
||||
|
||||
finish_desktop_task("t1")
|
||||
|
||||
root = tmp_path / "hermes-home" / "telemetry" / "shared_metrics"
|
||||
assert list((root / "outbox").glob("*.json")) == []
|
||||
with sqlite3.connect(root / "metrics.sqlite3") as connection:
|
||||
[package_count] = connection.execute(
|
||||
"SELECT COUNT(*) FROM package_outbox"
|
||||
).fetchone()
|
||||
assert package_count == 0
|
||||
|
||||
finish_desktop_task("t2")
|
||||
|
||||
[package_path] = list((root / "outbox").glob("*.json"))
|
||||
package = json.loads(package_path.read_text(encoding="utf-8"))
|
||||
metrics = {metric["name"]: metric for metric in package["metrics"]}
|
||||
assert metrics["hermes.task_run.started"]["value"] == 2
|
||||
assert metrics["hermes.task_run.finished"]["value"] == 2
|
||||
assert flush_attempts == 2
|
||||
assert "Hermes shared-metrics task flush failed" in caplog.text
|
||||
|
||||
|
||||
def test_task_ownership_survives_session_id_rotation(direct_runtime):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue