ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 11 Packages Projects Releases Wiki Activity

Merge branch 'simp/r3-22-C' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:58:12 -07:00
parent 50c49fa45c 69c419e525
commit 881a806e3d
7 changed files with 735 additions and 1414 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/observability/__init__.py
+4 -10
View File
@@ -12,7 +12,10 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
"""Dispatch a Hermes lifecycle event to built-in observability features."""
from . import relay_shared_metrics
_safe_observe(relay_shared_metrics.observe_lifecycle, hook_name, kwargs)
try:
relay_shared_metrics.observe_lifecycle(hook_name, **kwargs)
except Exception:
logger.warning("Built-in observability hook failed: %s", hook_name, exc_info=True)
def handles_hook(hook_name: str) -> bool:
@@ -20,12 +23,3 @@ def handles_hook(hook_name: str) -> bool:
from . import relay_shared_metrics
return relay_shared_metrics.handles_hook(hook_name)
def _safe_observe(callback: Any, hook_name: str, kwargs: dict[str, Any]) -> None:
try:
callback(hook_name, **kwargs)
except Exception:
logger.warning(
"Built-in observability hook failed: %s", hook_name, exc_info=True
)
hermes_cli/observability/relay_shared_metrics.py
+316 -637
View File
File diff suppressed because it is too large Load Diff
hermes_cli/observability/shared_metrics.py
+167 -363
View File
@@ -32,6 +32,8 @@ _SCHEMA_BUSY_TIMEOUT_MS = 5_000
_LOCAL_HISTORY_RETENTION_DAYS = 30
_ACTIVE_INSTALL_STATE_KEY = "client_active_recorded_at"
_ACTIVE_INSTALL_INTERVAL = timedelta(hours=24)
# Column order of the client resource in every counter_aggregates statement.
_RESOURCE_COLUMNS = ("hermes_version", "os_family", "architecture", "install_method")
logger = logging.getLogger(__name__)
@@ -44,6 +46,25 @@ def _isoformat(value: datetime) -> str:
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
def _compact_json(value: Any) -> str:
return json.dumps(value, sort_keys=True, separators=(",", ":"))
def _row_resource(row: sqlite3.Row) -> dict[str, str]:
return {column: row[column] for column in _RESOURCE_COLUMNS}
def _ensure_private(path: Path, mode: int) -> None:
if mode == 0o700:
path.mkdir(parents=True, exist_ok=True, mode=mode)
else:
path.touch(mode=mode, exist_ok=True)
try:
path.chmod(mode)
except OSError:
pass
class SharedMetricsStore:
"""Persist allowlisted counters and export immutable delta packages."""
@@ -55,16 +76,12 @@ class SharedMetricsStore:
root = get_hermes_home() / "telemetry" / "shared_metrics"
self.database_path = database_path or root / "metrics.sqlite3"
self.outbox_directory = outbox_directory or root / "outbox"
self._ensure_private_directory(self.database_path.parent)
self._ensure_private_directory(self.outbox_directory)
self._ensure_private_file(self.database_path)
_ensure_private(self.database_path.parent, 0o700)
_ensure_private(self.outbox_directory, 0o700)
_ensure_private(self.database_path, 0o600)
self._ensure_schema()
def record_model_call(
self,
dimensions: dict[str, str],
resource: dict[str, str],
) -> None:
def record_model_call(self, dimensions: dict[str, str], resource: dict[str, str]) -> None:
"""Increment the terminal model-call counter for the current UTC day."""
self.record_counter(MODEL_ROUTE_METRIC, dimensions, resource)
@@ -73,72 +90,50 @@ class SharedMetricsStore:
dimensions: dict[str, str] = {}
self._validate_counter(CLIENT_ACTIVE_METRIC, dimensions, resource)
now = _utc_now()
with self._connection() as connection:
with write_txn(connection):
row = connection.execute(
"SELECT value FROM telemetry_state WHERE key = ?",
(_ACTIVE_INSTALL_STATE_KEY,),
).fetchone()
if row is not None:
last_recorded = self._parse_state_timestamp(row["value"])
if last_recorded is not None and last_recorded > now:
# A wall-clock correction must not suppress activity until
# the stale future timestamp plus another full interval.
connection.execute(
"""
UPDATE telemetry_state
SET value = ?
WHERE key = ?
""",
(_isoformat(now), _ACTIVE_INSTALL_STATE_KEY),
)
return False
if (
last_recorded is not None
and now < last_recorded + _ACTIVE_INSTALL_INTERVAL
):
return False
with self._write() as connection:
row = connection.execute(
"SELECT value FROM telemetry_state WHERE key = ?",
(_ACTIVE_INSTALL_STATE_KEY,),
).fetchone()
last_recorded = self._parse_state_timestamp(row["value"]) if row is not None else None
if last_recorded is not None:
if last_recorded > now:
# A wall-clock correction must not suppress activity until the
# stale future timestamp plus another full interval.
connection.execute(
"UPDATE telemetry_state SET value = ? WHERE key = ?",
(_isoformat(now), _ACTIVE_INSTALL_STATE_KEY),
)
return False
if now < last_recorded + _ACTIVE_INSTALL_INTERVAL:
return False
self._install_id(connection)
self._record_counter_in_transaction(
connection,
CLIENT_ACTIVE_METRIC,
dimensions,
resource,
period_start=now.date().isoformat(),
)
connection.execute(
"""
INSERT INTO telemetry_state(key, value)
VALUES (?, ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""",
(_ACTIVE_INSTALL_STATE_KEY, _isoformat(now)),
)
self._install_id(connection)
self._record_counter_in_transaction(
connection, CLIENT_ACTIVE_METRIC, dimensions, resource,
period_start=now.date().isoformat(),
)
connection.execute(
"INSERT INTO telemetry_state(key, value) VALUES (?, ?)"
" ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(_ACTIVE_INSTALL_STATE_KEY, _isoformat(now)),
)
return True
def record_counter(
self,
metric_name: str,
dimensions: dict[str, str],
resource: dict[str, str],
self, metric_name: str, dimensions: dict[str, str], resource: dict[str, str]
) -> None:
"""Increment one allowlisted counter for the current UTC day."""
self._validate_counter(metric_name, dimensions, resource)
with self._connection() as connection:
self._record_counter_in_transaction(
connection,
metric_name,
dimensions,
resource,
connection, metric_name, dimensions, resource,
period_start=_utc_now().date().isoformat(),
)
@staticmethod
def _validate_counter(
metric_name: str,
dimensions: dict[str, str],
resource: dict[str, str],
metric_name: str, dimensions: dict[str, str], resource: dict[str, str]
) -> None:
if metric_name not in COUNTER_METRICS:
raise ValueError(f"Unsupported shared metric: {metric_name}")
@@ -156,50 +151,29 @@ class SharedMetricsStore:
*,
period_start: str,
) -> None:
dimensions_json = json.dumps(
dimensions,
sort_keys=True,
separators=(",", ":"),
)
connection.execute(
"""
INSERT INTO counter_aggregates(
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
period_start, metric_name, hermes_version, os_family, architecture,
install_method, dimensions_json, value, packaged_value
) VALUES (?, ?, ?, ?, ?, ?, ?, 1, 0)
ON CONFLICT(
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json
period_start, metric_name, hermes_version, os_family, architecture,
install_method, dimensions_json
)
DO UPDATE SET value = value + 1
""",
(
period_start,
metric_name,
resource["hermes_version"],
resource["os_family"],
resource["architecture"],
resource["install_method"],
dimensions_json,
*(resource[column] for column in _RESOURCE_COLUMNS),
_compact_json(dimensions),
),
)
def create_and_export_package(self) -> list[Path]:
"""Commit one pending delta package, then atomically export the outbox."""
pending_periods = self._pending_period_count()
for _ in range(pending_periods):
for _ in range(self._pending_period_count()):
if self._create_package() is None:
break
return self._export_and_prune()
@@ -214,10 +188,7 @@ class SharedMetricsStore:
try:
self._prune_expired_history()
except Exception:
logger.warning(
"Unable to prune expired shared-metrics history",
exc_info=True,
)
logger.warning("Unable to prune expired shared-metrics history", exc_info=True)
return exported
def counter_snapshot(self) -> list[dict[str, Any]]:
@@ -226,36 +197,19 @@ class SharedMetricsStore:
rows = connection.execute(
"""
SELECT
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
period_start, metric_name, hermes_version, os_family, architecture,
install_method, dimensions_json, value, packaged_value
FROM counter_aggregates
ORDER BY
period_start,
hermes_version,
os_family,
architecture,
install_method,
metric_name,
dimensions_json
period_start, hermes_version, os_family, architecture, install_method,
metric_name, dimensions_json
"""
).fetchall()
return [
{
"period_start": row["period_start"],
"metric_name": row["metric_name"],
"resource": {
"hermes_version": row["hermes_version"],
"os_family": row["os_family"],
"architecture": row["architecture"],
"install_method": row["install_method"],
},
"resource": _row_resource(row),
"dimensions": json.loads(row["dimensions_json"]),
"value": row["value"],
"packaged_value": row["packaged_value"],
@@ -265,14 +219,9 @@ class SharedMetricsStore:
@contextmanager
def _connection(
self,
*,
busy_timeout_ms: int = _BUSY_TIMEOUT_MS,
self, *, busy_timeout_ms: int = _BUSY_TIMEOUT_MS
) -> Iterator[sqlite3.Connection]:
connection = sqlite3.connect(
self.database_path,
timeout=busy_timeout_ms / 1000,
)
connection = sqlite3.connect(self.database_path, timeout=busy_timeout_ms / 1000)
try:
connection.row_factory = sqlite3.Row
connection.execute(f"PRAGMA busy_timeout={busy_timeout_ms}")
@@ -281,27 +230,17 @@ class SharedMetricsStore:
finally:
connection.close()
@staticmethod
def _ensure_private_directory(path: Path) -> None:
path.mkdir(parents=True, exist_ok=True, mode=0o700)
try:
path.chmod(0o700)
except OSError:
pass
@staticmethod
def _ensure_private_file(path: Path) -> None:
path.touch(mode=0o600, exist_ok=True)
try:
path.chmod(0o600)
except OSError:
pass
@contextmanager
def _write(self, *, busy_timeout_ms: int = _BUSY_TIMEOUT_MS) -> Iterator[sqlite3.Connection]:
"""A connection inside one write transaction."""
with self._connection(busy_timeout_ms=busy_timeout_ms) as connection:
with write_txn(connection):
yield connection
def _ensure_schema(self) -> None:
with self._connection(busy_timeout_ms=_SCHEMA_BUSY_TIMEOUT_MS) as connection:
# Serialize first-run creation and upgrades across Hermes processes.
with write_txn(connection):
self._ensure_schema_in_transaction(connection)
# Serialize first-run creation and upgrades across Hermes processes.
with self._write(busy_timeout_ms=_SCHEMA_BUSY_TIMEOUT_MS) as connection:
self._ensure_schema_in_transaction(connection)
@staticmethod
def _ensure_schema_in_transaction(connection: sqlite3.Connection) -> None:
@@ -340,11 +279,8 @@ class SharedMetricsStore:
SharedMetricsStore._add_send_columns(connection)
SharedMetricsStore._add_consent_tables(connection)
connection.execute(
"""
INSERT INTO telemetry_state(key, value)
VALUES ('schema_version', ?)
ON CONFLICT(key) DO UPDATE SET value = excluded.value
""",
"INSERT INTO telemetry_state(key, value) VALUES ('schema_version', ?)"
" ON CONFLICT(key) DO UPDATE SET value = excluded.value",
(_STORE_SCHEMA_VERSION,),
)
@@ -352,11 +288,10 @@ class SharedMetricsStore:
def _add_send_columns(connection: sqlite3.Connection) -> None:
"""Add transmission bookkeeping to ``package_outbox``, idempotently.
The columns are ADDITIVE and nullable, and the schema version is deliberately NOT bumped:
``_ensure_schema_in_transaction`` raises on unknown versions with no forward-compat branch,
so a bump would hard-fail an older Hermes (second profile, rollback) against the same
database. Old readers select named columns, never ``SELECT *``, so extra columns are
invisible to them.
Additive and nullable; the schema version is deliberately NOT bumped because
``_ensure_schema_in_transaction`` raises on unknown versions, so a bump would
hard-fail an older Hermes (second profile, rollback) sharing the database. Old
readers select named columns, never ``SELECT *``.
"""
existing = {
str(row["name"])
@@ -371,17 +306,12 @@ class SharedMetricsStore:
# Earliest next attempt; enforces backoff across process restarts.
("next_attempt_at", "TEXT"),
("last_error", "TEXT"),
# The identifier actually transmitted, frozen on the first
# attempt so retries stay byte-identical. Since the 2026-08-27
# product decision this is the stable install_id itself.
# Only the ~36-byte id is stored: the body is recomputed from
# payload_json, whose serialisation is deterministic.
# The install_id actually transmitted, frozen on the first attempt so
# retries stay byte-identical; the body is recomputed from payload_json.
("sent_install_id", "TEXT"),
# NULL until first claimed; rewritten on every claim. Settlement
# and the pre-POST revalidation are compare-and-set on this, so a
# claimant whose lease lapsed loses authority the moment another
# process reclaims (PR-review finding: without it, a suspended
# sender resuming after a reclaim double-POSTs the package).
# Rewritten on every claim. Settlement and the pre-POST revalidation are
# compare-and-set on it, so a claimant whose lease lapsed loses authority
# the moment another process reclaims (else a suspended sender double-POSTs).
("claim_token", "TEXT"),
):
if column not in existing:
@@ -391,15 +321,11 @@ class SharedMetricsStore:
@staticmethod
def _add_consent_tables(connection: sqlite3.Connection) -> None:
"""Create the consent-window tables, idempotently.
"""Create the consent-window tables, idempotently (additive; version not bumped).
Additive like ``_add_send_columns`` — the schema version is deliberately NOT bumped, and old
readers never touch these tables.
``send_consent_windows`` records consent as explicit intervals rather than a moving day-
stamp: a window is opened when send consent is observed, heartbeat-confirmed on every later
observation, and closed at the LAST CONFIRMED moment (never "now") when consent is observed
withdrawn.
``send_consent_windows`` records consent as explicit intervals rather than a moving
day-stamp: opened when send consent is observed, heartbeat-confirmed on every later
observation, closed at the LAST CONFIRMED moment (never "now") on withdrawal.
"""
connection.execute(
"""
@@ -434,13 +360,8 @@ class SharedMetricsStore:
value INTEGER NOT NULL,
packaged_value INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json
period_start, metric_name, hermes_version, os_family, architecture,
install_method, dimensions_json
)
)
"""
@@ -455,45 +376,26 @@ class SharedMetricsStore:
connection.execute(
"""
INSERT INTO counter_aggregates(
period_start,
metric_name,
hermes_version,
os_family,
architecture,
install_method,
dimensions_json,
value,
packaged_value
period_start, metric_name, hermes_version, os_family, architecture,
install_method, dimensions_json, value, packaged_value
)
SELECT
period_start,
metric_name,
hermes_version,
'unknown',
'unknown',
'unknown',
dimensions_json,
value,
packaged_value
period_start, metric_name, hermes_version, 'unknown', 'unknown', 'unknown',
dimensions_json, value, packaged_value
FROM counter_aggregates_v1
"""
)
connection.execute("DROP TABLE counter_aggregates_v1")
def _install_id(self, connection: sqlite3.Connection) -> str:
row = connection.execute(
"SELECT value FROM telemetry_state WHERE key = 'install_id'"
).fetchone()
if row is not None:
return str(row["value"])
candidate = str(uuid.uuid4())
connection.execute(
"INSERT OR IGNORE INTO telemetry_state(key, value) VALUES ('install_id', ?)",
(candidate,),
)
row = connection.execute(
"SELECT value FROM telemetry_state WHERE key = 'install_id'"
).fetchone()
query = "SELECT value FROM telemetry_state WHERE key = 'install_id'"
row = connection.execute(query).fetchone()
if row is None:
connection.execute(
"INSERT OR IGNORE INTO telemetry_state(key, value) VALUES ('install_id', ?)",
(str(uuid.uuid4()),),
)
row = connection.execute(query).fetchone()
if row is None:
raise RuntimeError("Unable to create the shared-metrics install identity")
return str(row["value"])
@@ -504,9 +406,7 @@ class SharedMetricsStore:
parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
except (TypeError, ValueError):
return None
if parsed.tzinfo is None:
return None
return parsed.astimezone(timezone.utc)
return None if parsed.tzinfo is None else parsed.astimezone(timezone.utc)
def _pending_period_count(self) -> int:
with self._connection() as connection:
@@ -514,20 +414,10 @@ class SharedMetricsStore:
"""
SELECT COUNT(*) AS period_count
FROM (
SELECT
period_start,
hermes_version,
os_family,
architecture,
install_method
SELECT period_start, hermes_version, os_family, architecture, install_method
FROM counter_aggregates
WHERE value > packaged_value
GROUP BY
period_start,
hermes_version,
os_family,
architecture,
install_method
GROUP BY period_start, hermes_version, os_family, architecture, install_method
)
"""
).fetchone()
@@ -535,127 +425,80 @@ class SharedMetricsStore:
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
with self._write() as 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:
with write_txn(connection):
return self._create_package_in_transaction(connection, now)
with self._write() as connection:
return self._create_package_in_transaction(connection, _utc_now())
def _create_package_in_transaction(
self,
connection: sqlite3.Connection,
now: datetime,
self, connection: sqlite3.Connection, now: datetime
) -> dict[str, Any] | None:
period_row = connection.execute(
"""
SELECT
period_start,
hermes_version,
os_family,
architecture,
install_method
SELECT period_start, hermes_version, os_family, architecture, install_method
FROM counter_aggregates
WHERE value > packaged_value
ORDER BY
period_start,
hermes_version,
os_family,
architecture,
install_method
ORDER BY period_start, hermes_version, os_family, architecture, install_method
LIMIT 1
"""
).fetchone()
period_value = period_row["period_start"] if period_row is not None else None
if not period_value:
if period_row is None or not period_row["period_start"]:
return None
period_value = period_row["period_start"]
resource = _row_resource(period_row)
resource_values = tuple(resource[column] for column in _RESOURCE_COLUMNS)
rows = connection.execute(
"""
SELECT metric_name, dimensions_json, value, packaged_value
FROM counter_aggregates
WHERE period_start = ?
AND hermes_version = ?
AND os_family = ?
AND architecture = ?
AND install_method = ?
AND value > packaged_value
WHERE period_start = ? AND hermes_version = ? AND os_family = ?
AND architecture = ? AND install_method = ? AND value > packaged_value
ORDER BY metric_name, dimensions_json
""",
(
period_value,
period_row["hermes_version"],
period_row["os_family"],
period_row["architecture"],
period_row["install_method"],
),
(period_value, *resource_values),
).fetchall()
period_start = datetime.fromisoformat(str(period_value)).replace(
tzinfo=timezone.utc
)
period_end = period_start + timedelta(days=1)
package_id = str(uuid.uuid4())
resource = {
"hermes_version": period_row["hermes_version"],
"os_family": period_row["os_family"],
"architecture": period_row["architecture"],
"install_method": period_row["install_method"],
}
period_start = datetime.fromisoformat(str(period_value)).replace(tzinfo=timezone.utc)
if not client_resource_is_valid(resource):
raise ValueError("Unsupported shared-metrics client resource")
payload = {
"schema_version": _PACKAGE_SCHEMA_VERSION,
"package_id": package_id,
"package_id": str(uuid.uuid4()),
"install_id": self._install_id(connection),
"period_start": _isoformat(period_start),
"period_end": _isoformat(period_end),
"period_end": _isoformat(period_start + timedelta(days=1)),
"generated_at": _isoformat(now),
"resource": resource,
"metrics": [self._package_metric(row) for row in rows],
}
payload_json = json.dumps(
payload,
sort_keys=True,
separators=(",", ":"),
)
connection.execute(
"""
INSERT INTO package_outbox(
package_id,
period_start,
period_end,
payload_json,
created_at
package_id, period_start, period_end, payload_json, created_at
) VALUES (?, ?, ?, ?, ?)
""",
(
package_id,
payload["package_id"],
payload["period_start"],
payload["period_end"],
payload_json,
_compact_json(payload),
payload["generated_at"],
),
)
# Advance the data high-water mark. This is the ONLY writer of the
# 'data' mark: it clamps consent-window opens so a rolled-back clock
# can never open a window underneath packages that already exist.
# Advance the data high-water mark. This is the ONLY writer of the 'data' mark:
# it clamps consent-window opens so a rolled-back clock can never open a window
# underneath packages that already exist.
connection.execute(
"""
INSERT INTO consent_marks(name, stamp) VALUES ('data', ?)
@@ -668,23 +511,11 @@ class SharedMetricsStore:
"""
UPDATE counter_aggregates
SET packaged_value = value
WHERE period_start = ?
AND metric_name = ?
AND hermes_version = ?
AND os_family = ?
AND architecture = ?
AND install_method = ?
WHERE period_start = ? AND metric_name = ? AND hermes_version = ?
AND os_family = ? AND architecture = ? AND install_method = ?
AND dimensions_json = ?
""",
(
period_value,
row["metric_name"],
period_row["hermes_version"],
period_row["os_family"],
period_row["architecture"],
period_row["install_method"],
row["dimensions_json"],
),
(period_value, row["metric_name"], *resource_values, row["dimensions_json"]),
)
return payload
@@ -706,12 +537,8 @@ class SharedMetricsStore:
def _export_pending_packages(self) -> list[Path]:
with self._connection() as connection:
rows = connection.execute(
"""
SELECT package_id, payload_json
FROM package_outbox
WHERE exported_at IS NULL
ORDER BY created_at, package_id
"""
"SELECT package_id, payload_json FROM package_outbox"
" WHERE exported_at IS NULL ORDER BY created_at, package_id"
).fetchall()
exported: list[Path] = []
@@ -719,19 +546,12 @@ class SharedMetricsStore:
package_id = str(row["package_id"])
path = self.outbox_directory / f"{package_id}.json"
atomic_json_write(
path,
json.loads(row["payload_json"]),
indent=2,
sort_keys=True,
mode=0o600,
path, json.loads(row["payload_json"]), indent=2, sort_keys=True, mode=0o600
)
with self._connection() as connection:
connection.execute(
"""
UPDATE package_outbox
SET exported_at = ?
WHERE package_id = ? AND exported_at IS NULL
""",
"UPDATE package_outbox SET exported_at = ?"
" WHERE package_id = ? AND exported_at IS NULL",
(_isoformat(_utc_now()), package_id),
)
exported.append(path)
@@ -739,20 +559,13 @@ class SharedMetricsStore:
def _prune_expired_history(self, *, now: datetime | None = None) -> None:
"""Remove exported local history after the bounded retention window."""
cutoff = (now or _utc_now()) - timedelta(
days=_LOCAL_HISTORY_RETENTION_DAYS
)
cutoff = (now or _utc_now()) - timedelta(days=_LOCAL_HISTORY_RETENTION_DAYS)
cutoff_timestamp = _isoformat(cutoff)
cutoff_period = cutoff.date().isoformat()
with self._connection() as connection:
rows = connection.execute(
"""
SELECT package_id
FROM package_outbox
WHERE exported_at IS NOT NULL
AND exported_at < ?
ORDER BY exported_at, package_id
""",
"SELECT package_id FROM package_outbox"
" WHERE exported_at IS NOT NULL AND exported_at < ?"
" ORDER BY exported_at, package_id",
(cutoff_timestamp,),
).fetchall()
@@ -760,32 +573,23 @@ class SharedMetricsStore:
for row in rows:
package_id = str(row["package_id"])
try:
(self.outbox_directory / f"{package_id}.json").unlink(
missing_ok=True
)
(self.outbox_directory / f"{package_id}.json").unlink(missing_ok=True)
except OSError:
logger.warning(
"Unable to prune expired shared-metrics package %s",
package_id,
exc_info=True,
"Unable to prune expired shared-metrics package %s", package_id, exc_info=True
)
continue
removable_package_ids.append(package_id)
with self._connection() as connection:
with write_txn(connection):
for package_id in removable_package_ids:
connection.execute(
"""
DELETE FROM package_outbox
WHERE package_id = ?
AND exported_at IS NOT NULL
AND exported_at < ?
""",
(package_id, cutoff_timestamp),
)
with self._write() as connection:
for package_id in removable_package_ids:
connection.execute(
"""
"DELETE FROM package_outbox"
" WHERE package_id = ? AND exported_at IS NOT NULL AND exported_at < ?",
(package_id, cutoff_timestamp),
)
connection.execute(
"""
DELETE FROM counter_aggregates
WHERE period_start < ?
AND value = packaged_value
@@ -797,5 +601,5 @@ class SharedMetricsStore:
= counter_aggregates.period_start
)
""",
(cutoff_period,),
)
(cutoff.date().isoformat(),),
)
hermes_cli/observability/shared_metrics_contract.py
+23 -31
View File
@@ -100,6 +100,7 @@ _ARCHITECTURE_ALIASES = {
"aarch64": "arm64", "arm64": "arm64",
"i386": "x86", "i486": "x86", "i586": "x86", "i686": "x86", "x86": "x86",
}
_OS_FAMILIES = {"darwin": "macos", "linux": "linux", "macos": "macos", "windows": "windows"}
def _norm(value: Any) -> str:
@@ -112,9 +113,7 @@ def _allowlisted(normalized: str, allowed: frozenset[str]) -> str:
def client_os_family(value: Any) -> str:
"""Map a platform system name to the shared-metrics OS taxonomy."""
return {"darwin": "macos", "linux": "linux", "macos": "macos", "windows": "windows"}.get(
_norm(value), "unknown"
)
return _OS_FAMILIES.get(_norm(value), "unknown")
def client_architecture(value: Any) -> str:
@@ -279,24 +278,25 @@ def _event_shape_matches(
"""Match the coarse Relay event shape.
A ``str`` expectation compares against the stringified attribute; ``None`` requires the
attribute itself to be ``None``; ``_ANY`` skips the check. ``category_profile`` compares
with plain equality (marks carry ``None``, tool scopes ``{}``).
attribute itself to be ``None``; ``_ANY`` skips the check. Anything else (the
``category_profile`` dict) compares with plain equality.
"""
if _event_text(event, "kind") != kind:
return False
for attr, expected, as_text in (
("name", name, True),
("category", category, True),
("scope_category", scope_category, True),
("category_profile", category_profile, False),
):
expectations = {
"name": name, "category": category, "scope_category": scope_category,
"category_profile": category_profile,
}
for attr, expected in expectations.items():
if expected is _ANY:
continue
actual = getattr(event, attr, None)
if expected is None:
if actual is not None:
return False
elif (_event_text(event, attr) if as_text else actual) != expected:
if isinstance(expected, str):
matched = _event_text(event, attr) == expected
elif expected is None:
matched = getattr(event, attr, None) is None
else:
matched = getattr(event, attr, None) == expected
if not matched:
return False
return True
@@ -453,6 +453,7 @@ _SURFACE_ALIASES = {
"api_server": "api",
**dict.fromkeys(("cron", "scheduler", "scheduled"), "scheduled_task"),
}
_KNOWN_GATEWAY_PLATFORMS = frozenset({"discord", "email", "slack", "telegram", "teams", "whatsapp"})
def execution_surface(kwargs: dict[str, Any]) -> str:
@@ -469,7 +470,7 @@ def execution_surface(kwargs: dict[str, Any]) -> str:
return "gateway"
except Exception:
pass
if value in {"discord", "email", "slack", "telegram", "teams", "whatsapp"}:
if value in _KNOWN_GATEWAY_PLATFORMS:
return "gateway"
return "unknown" if value == "unknown" else "other"
@@ -477,10 +478,7 @@ def execution_surface(kwargs: dict[str, Any]) -> str:
def task_start_fields(kwargs: dict[str, Any]) -> dict[str, str]:
"""Build the bounded fields recorded on a task scope start event."""
surface = execution_surface(kwargs)
return {
"entrypoint": task_entrypoint(kwargs, surface),
"execution_surface": surface,
}
return {"entrypoint": task_entrypoint(kwargs, surface), "execution_surface": surface}
_SURFACE_ENTRYPOINTS = {
@@ -495,10 +493,9 @@ def task_entrypoint(kwargs: dict[str, Any], surface: str | None = None) -> str:
declared = _norm(kwargs.get("entrypoint"))
if declared in TASK_ENTRYPOINTS:
return declared
resolved_surface = surface or execution_surface(kwargs)
if kwargs.get("parent_task_id") or kwargs.get("parent_session_id"):
return "delegated"
return _SURFACE_ENTRYPOINTS.get(resolved_surface, "other")
return _SURFACE_ENTRYPOINTS.get(surface or execution_surface(kwargs), "other")
def task_terminal_fields(
@@ -524,7 +521,7 @@ def task_terminal_fields(
def task_terminal_state(kwargs: dict[str, Any]) -> tuple[str, str, str]:
"""Map Hermes terminal state to bounded task outcome dimensions."""
"""Map Hermes terminal state to bounded (outcome, end_reason, termination)."""
reason = _norm(kwargs.get("turn_exit_reason"))
if kwargs.get("interrupted") or "interrupt" in reason or "cancel" in reason:
return "cancelled", "user_cancelled", "user_cancelled"
@@ -558,10 +555,7 @@ _LATENCY_THRESHOLDS = (
def _bucket(value: float, thresholds: tuple[tuple[float, str], ...], last: str) -> str:
for upper, label in thresholds:
if value < upper:
return label
return last
return next((label for upper, label in thresholds if value < upper), last)
def duration_bucket(duration_ms: int) -> str:
@@ -636,9 +630,7 @@ def tool_terminal_fields(
) -> dict[str, str]:
"""Build one bounded tool-call terminal payload."""
return {
"approval_outcome": (
approval_outcome if approval_outcome in TOOL_APPROVAL_OUTCOMES else "unknown"
),
"approval_outcome": _allowlisted(approval_outcome, TOOL_APPROVAL_OUTCOMES),
"latency_bucket": tool_latency_bucket(
kwargs.get("duration_ms"), fallback_duration_ms=fallback_duration_ms
),
hermes_cli/observability/shared_metrics_send_config.py
+9 -22
View File
@@ -8,14 +8,10 @@ from urllib.parse import urlparse
logger = logging.getLogger(__name__)
#: Production ingest endpoint. Overridable through config only.
#:
#: Deliberately NOT overridable by an environment variable: AGENTS.md reserves
#: HERMES_* env vars for secrets, and a behavioural override here would be a
#: consent hazard — a user who agreed to send metrics to Nous could have them
#: silently redirected to any host by an inherited variable, with nothing
#: visible in their config to show it. Tests and the staging E2E write this
#: key into a throwaway profile instead.
#: Production ingest endpoint. Overridable through config only — deliberately NOT by an
#: environment variable: AGENTS.md reserves HERMES_* for secrets, and an inherited variable
#: could silently redirect consented metrics to any host with nothing visible in config.
#: Tests and the staging E2E write this key into a throwaway profile instead.
DEFAULT_ENDPOINT = "https://telemetry.nousresearch.com/v1/telemetry"
_LOCAL_HOSTS = frozenset({"localhost", "127.0.0.1", "::1", "[::1]"})
@@ -38,28 +34,20 @@ class SendConfig:
def _endpoint_is_safe(endpoint: str) -> bool:
"""Reject plaintext destinations unless they are loopback.
Telemetry must not leave a machine in clear text because of a typo in a config file. Loopback
stays allowed so tests can use a local HTTP server.
"""
"""Reject plaintext destinations unless they are loopback (so tests can use local HTTP)."""
try:
parsed = urlparse(endpoint)
except ValueError:
return False
if parsed.scheme == "https":
return True
if parsed.scheme == "http":
return (parsed.hostname or "") in _LOCAL_HOSTS
return False
return parsed.scheme == "http" and (parsed.hostname or "") in _LOCAL_HOSTS
def resolve_send_config(config: dict | None) -> SendConfig:
"""Resolve transmission settings from config plus the environment.
"""Resolve transmission settings from config (endpoint: config > production default).
Endpoint precedence: config > production default.
``send`` is returned as False whenever transmission cannot legitimately happen, so callers never
``send`` is False whenever transmission cannot legitimately happen, so callers never
have to re-check the combination.
"""
global _warned_send_without_collection
@@ -74,8 +62,7 @@ def resolve_send_config(config: dict | None) -> SendConfig:
send_requested = shared.get("send") is True
if send_requested and not enabled:
# Loud, not silent: the user believes telemetry is being sent, and it
# never will be. Error level, once per process.
# Loud, not silent: the user believes telemetry is being sent, and it never will be.
if not _warned_send_without_collection:
_warned_send_without_collection = True
logger.error(
hermes_cli/observability/shared_metrics_sender.py
+199 -332
View File
@@ -1,10 +1,8 @@
"""Transmit exported shared-metrics packages to the Nous telemetry service.
Implements the sender side of the ingest contract (see the telemetry repo's ``CONTRACT.md``):
* ``202`` — durably stored. Mark sent. * ``400`` — permanently malformed. Never retry. * ``429`` —
keep, retry after ``Retry-After``. * ``5xx`` / timeout / connection error — keep, retry with
backoff.
Sender side of the ingest contract (telemetry repo ``CONTRACT.md``): ``202`` durably stored,
mark sent; ``400`` permanently malformed, never retry; ``429`` keep, retry after
``Retry-After``; ``5xx`` / timeout / connection error keep, retry with backoff.
"""
from __future__ import annotations
@@ -18,11 +16,14 @@ import time
import urllib.error
import urllib.request
import uuid
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from hermes_cli.sqlite_util import write_txn
from .shared_metrics import _isoformat, _utc_now
logger = logging.getLogger(__name__)
#: Contract recommends timing out at 30s and treating a timeout as retryable.
@@ -41,49 +42,40 @@ GZIP_THRESHOLD_BYTES = 4096
MAX_PACKAGES_PER_PASS = 20
#: How long a claimed row is held. The claim writes a LEASE INTO THE FUTURE:
#: selection requires `next_attempt_at <= now`, so for the length of the lease
#: no other process can take the package.
#:
#: This must exceed the worst case for ONE package — three 30s request
#: timeouts plus 1s+5s of backoff, about 96s — which is why packages are
#: claimed one at a time, immediately before being sent. An earlier revision
#: claimed up to 20 rows under a single shared lease; a full batch can legally
#: run ~1900s, so the later rows' leases expired while the pass still held
#: them in memory and another process re-sent them.
#: selection requires `next_attempt_at <= now`, so no other process can take the
#: package for the length of the lease. Must exceed the worst case for ONE package
#: (three 30s timeouts plus 1s+5s backoff, ~96s) — which is why packages are claimed
#: one at a time right before sending: a batch of 20 under one lease can legally run
#: ~1900s, so later rows' leases expired while still held and got re-sent.
_CLAIM_LEASE_SECONDS = 300
#: Floor applied after a pass fails to deliver, so a hard-down service is not
#: retried on every task completion.
_FAILURE_BACKOFF_SECONDS = 15 * 60
#: Statuses that are permanent per the ingest contract. Deliberately narrow:
#: 400 means the envelope is malformed and will never validate. 413 is added
#: because a package over the service's 1 MiB cap cannot shrink on retry.
#: Everything else — including 403 from the origin guard and 404 from a bad
#: path — is retried, because those are usually deployment or edge
#: Permanent statuses per the ingest contract. Deliberately narrow: 400 means the
#: envelope will never validate; 413 because a package over the 1 MiB cap cannot
#: shrink on retry. Everything else — including 403 from the origin guard and 404
#: from a bad path — is retried, since those are usually deployment/edge
#: misconfiguration that resolves without the package changing.
_PERMANENT_STATUSES = frozenset({400, 413})
#: Attempts after which a package is abandoned. Without a ceiling a
#: permanently-poisoned row is retried until 30-day retention deletes it —
#: measured at ~160 requests — which wastes the user's bandwidth and keeps a
#: doomed package at the head of the queue.
#: Attempts after which a package is abandoned. Without a ceiling a poisoned row
#: is retried until 30-day retention deletes it (~160 requests), wasting the user's
#: bandwidth and pinning a doomed package at the head of the queue.
MAX_SEND_ATTEMPTS = 25
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
def _isoformat(value: datetime) -> str:
return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z")
#: Maximum distance one reconcile call can advance the 'obs' mark. Honest heartbeats
#: arrive hours apart so the cap never binds normally; a machine off for months catches
#: up in a few hook fires (fail-closed latency only). It bounds FORWARD clock poison:
#: without it one glitched sample (NTP flap reading 2099) drags the mark — and every
#: window open and confirmation horizon — decades ahead, a refused-data leak.
MAX_OBS_ADVANCE_SECONDS = 30 * 24 * 3600
def _parse_stamp(value: str) -> datetime:
"""Parse a stamp this module itself wrote (Z-suffixed ISO-8601, UTC)."""
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(
timezone.utc
)
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
@dataclass
@@ -112,17 +104,11 @@ def _post(endpoint: str, payload: bytes, *, timeout: int) -> _Response:
}
body = payload
if len(payload) > GZIP_THRESHOLD_BYTES:
# mtime=0: gzip embeds a timestamp by default, which would make two
# sends of one package differ on the wire. The service decompresses
# before storing so it would not change what lands in S3, but a
# deterministic body keeps "a resend is byte-identical" true at the
# transport layer too, and makes the property testable.
# mtime=0 keeps two sends of one package byte-identical on the wire too.
body = gzip.compress(payload, mtime=0)
headers["Content-Encoding"] = "gzip"
request = urllib.request.Request(
endpoint, data=body, headers=headers, method="POST"
)
request = urllib.request.Request(endpoint, data=body, headers=headers, method="POST")
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
return _Response(
@@ -143,25 +129,13 @@ def _retry_after_seconds(value: str | None, default: int) -> int:
if not value:
return default
try:
# Contract sends seconds. Clamp so a hostile or bogus value cannot
# park a package for years, and never go below one second.
# Contract sends seconds. Clamp so a bogus value cannot park a package for
# years, and never go below one second.
return max(1, min(int(float(value)), 86_400))
except (TypeError, ValueError):
return default
#: Maximum distance one reconcile call can advance the 'obs' mark. Honest
#: heartbeats arrive hours apart at most, so the cap never binds in normal
#: operation; a machine legitimately off for months catches up in a few
#: hook fires (fail-closed latency only). What it bounds is FORWARD clock
#: poison: without it, a single glitched sample (NTP flap reading 2099)
#: permanently drags the mark — and with it every window open and every
#: confirmation horizon — decades ahead, which round 6 reproduced as a
#: refused-data leak. Capped, one insane sample moves the mark at most
#: this far, and real time overtakes it again.
MAX_OBS_ADVANCE_SECONDS = 30 * 24 * 3600
def reconcile_send_consent(
connection: sqlite3.Connection,
send_enabled: bool,
@@ -170,9 +144,8 @@ def reconcile_send_consent(
) -> None:
"""Reconcile the consent-window table with the observed config state.
THE ONLY writer of consent state. Must run inside a write transaction. A pure function of
(config, now, store): call it from anywhere, any number of times, in any order — the resulting
windows are the same.
THE ONLY writer of consent state. Must run inside a write transaction. A pure function
of (config, now, store): call it from anywhere, any number of times, in any order.
"""
stamp = _isoformat(now or _utc_now())
raw_stamp = stamp # pre-cap observation time, used to clamp closes
@@ -181,8 +154,7 @@ def reconcile_send_consent(
).fetchone()
if previous_obs is not None:
ceiling = _isoformat(
_parse_stamp(str(previous_obs[0]))
+ timedelta(seconds=MAX_OBS_ADVANCE_SECONDS)
_parse_stamp(str(previous_obs[0])) + timedelta(seconds=MAX_OBS_ADVANCE_SECONDS)
)
stamp = min(stamp, ceiling)
connection.execute(
@@ -192,9 +164,7 @@ def reconcile_send_consent(
""",
(stamp,),
)
marks = dict(
connection.execute("SELECT name, stamp FROM consent_marks").fetchall()
)
marks = dict(connection.execute("SELECT name, stamp FROM consent_marks").fetchall())
obs = marks["obs"] # >= stamp; immune to clock rollback
data = marks.get("data")
@@ -218,17 +188,13 @@ def reconcile_send_consent(
(obs, open_row[0]),
)
elif open_row is not None:
# Close at the last CONFIRMED moment, but never after the closing
# observation's own raw stamp. The two clamps serve different
# adversaries and both are load-bearing:
# - min with last_confirmed_at: an unobserved gap (machine off,
# hand-edited config) is never asserted as consented (v1's leak).
# - min with the RAW stamp (pre-cap, pre-MAX): if last_confirmed_at
# was poisoned by a glitched-forward sample, an honest clock at
# revoke time pulls the close back to the true revoke moment, so
# the refused era that follows falls OUTSIDE the closed window
# (round 6's D1 leak). A rolled-back clock at close time only
# closes EARLIER — fail-closed.
# Close at the last CONFIRMED moment, but never after the closing observation's
# own raw stamp. Both clamps are load-bearing: min with last_confirmed_at means an
# unobserved gap (machine off, hand-edited config) is never asserted as consented;
# min with the RAW (pre-cap, pre-MAX) stamp means a last_confirmed_at poisoned by a
# glitched-forward sample is pulled back to the true revoke moment by an honest
# clock, so the refused era falls OUTSIDE the window. A rolled-back clock only
# closes EARLIER — fail-closed.
connection.execute(
"UPDATE send_consent_windows"
" SET closed_at = MIN(last_confirmed_at, ?)"
@@ -237,10 +203,10 @@ def reconcile_send_consent(
)
#: Claim-time consent predicate: the package's period must fall entirely
#: inside SOME recorded consent window. An open window vouches only up to its
#: last confirmed moment, so a package whose period runs past it waits for
#: the next reconcile heartbeat (fail-closed; released within one hook fire).
#: Claim-time consent predicate: the package's period must fall entirely inside SOME
#: recorded consent window. An open window vouches only up to its last confirmed moment,
#: so a package whose period runs past it waits for the next reconcile heartbeat
#: (fail-closed; released within one hook fire).
CONSENT_GATE_SQL = """EXISTS (
SELECT 1 FROM send_consent_windows w
WHERE package_outbox.period_start >= w.opened_at
@@ -270,36 +236,38 @@ class SharedMetricsSender:
self._sleep = sleep
self._now = now
self._max_attempts = max_attempts
# Called before every package. None disables the check for callers
# that have already established consent out of band (tests, E2E).
# Called before every package. None disables the check for callers that have
# already established consent out of band (tests, E2E).
self._consent_check = consent_check
@contextmanager
def _write(self):
"""One store connection inside a write transaction."""
with self._store._connection() as connection:
with write_txn(connection):
yield connection
# -- selection ---------------------------------------------------------
def _claim_next(self, now: datetime, seen: set[str]) -> dict | None:
"""Claim exactly ONE package, immediately before it is sent.
Claiming a batch up front fails: one shared lease would have to cover the whole pass, and 20
retrying packages can outlive any sane lease, so later rows expire and another process
re-sends them. ``seen`` (packages this pass finished with) is excluded IN SQL: with LIMIT 1,
returning None for a seen row would look like an empty queue and abandon everything behind
it, and rows can legitimately become eligible again mid-pass.
Claiming a batch up front fails: one shared lease would have to cover the whole
pass, and 20 retrying packages can outlive any sane lease. ``seen`` (packages this
pass finished with) is excluded IN SQL: with LIMIT 1, returning None for a seen row
would look like an empty queue and abandon everything behind it, and rows can
legitimately become eligible again mid-pass.
"""
with self._store._connection() as connection:
with write_txn(connection):
stamp = _isoformat(now)
lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
with self._write() as connection:
stamp = _isoformat(now)
lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
placeholders = ",".join("?" for _ in seen)
exclusion = (
f" AND package_id NOT IN ({placeholders})" if seen else ""
)
# Consent is a READ here — the claim must never mutate the
# window table. The old design's opt_in_period() call at this
# exact spot meant selecting a row could rewrite what was
# permitted to be sent (and did, under a rolled-back clock).
row = connection.execute(
f"""
placeholders = ",".join("?" for _ in seen)
exclusion = f" AND package_id NOT IN ({placeholders})" if seen else ""
# Consent is a READ here — the claim must never mutate the window table, or
# selecting a row could rewrite what is permitted to be sent.
row = connection.execute(
f"""
SELECT package_id, payload_json, sent_install_id
FROM package_outbox
WHERE exported_at IS NOT NULL
@@ -311,25 +279,20 @@ class SharedMetricsSender:
ORDER BY created_at, package_id
LIMIT 1
""",
(stamp, MAX_SEND_ATTEMPTS, *sorted(seen)),
).fetchone()
if row is None:
return None
(stamp, MAX_SEND_ATTEMPTS, *sorted(seen)),
).fetchone()
if row is None:
return None
package_id = str(row[0])
derived = row[2]
if not derived:
derived = self._freeze_identity(
connection, package_id, row[1], now
)
if derived is None:
# Unusable row, already marked rejected. Signal the
# caller to continue rather than stop.
return {"package_id": package_id, "skip": True}
package_id = str(row[0])
derived = row[2] or self._freeze_identity(connection, package_id, row[1])
if derived is None:
# Unusable row, already marked rejected. Tell the caller to continue.
return {"package_id": package_id, "skip": True}
token = str(uuid.uuid4())
connection.execute(
"""
token = str(uuid.uuid4())
connection.execute(
"""
UPDATE package_outbox
SET send_state = 'pending',
send_attempts = send_attempts + 1,
@@ -337,58 +300,49 @@ class SharedMetricsSender:
claim_token = ?
WHERE package_id = ?
""",
# Lease INTO THE FUTURE: selection requires
# next_attempt_at <= now, so no other process can take
# this row while it is in flight. Success or a real
# backoff overwrites it; if this process dies, it expires.
# The token is this claim's identity: a reclaim after
# expiry mints a new one, and every later write by THIS
# claimant is compare-and-set against it, so a lapsed
# claimant that resumes cannot settle or transmit.
(_isoformat(lease_until), token, package_id),
)
return {
"package_id": package_id,
"payload_json": str(row[1]),
"derived": str(derived),
"claim_token": token,
"skip": False,
}
# Lease INTO THE FUTURE (see _CLAIM_LEASE_SECONDS). The token is this
# claim's identity: a reclaim after expiry mints a new one, and every later
# write by THIS claimant is compare-and-set against it, so a lapsed claimant
# that resumes cannot settle or transmit.
(_isoformat(lease_until), token, package_id),
)
return {
"package_id": package_id,
"payload_json": str(row[1]),
"derived": str(derived),
"claim_token": token,
"skip": False,
}
@staticmethod
def _freeze_identity(
self,
connection: sqlite3.Connection,
package_id: str,
payload_json,
now: datetime,
connection: sqlite3.Connection, package_id: str, payload_json
) -> str | None:
"""Record the transmitted id on the row, or reject an unusable one.
"""Record the transmitted install_id on the row, or reject an unusable one.
What remains of "freezing" is the validation and the audit column: ``sent_install_id``
records exactly what the wire will carry, and rejecting unusable rows here rather than
raising matters because an exception rolls back the claim transaction and blocks every
``sent_install_id`` records exactly what the wire will carry. Rejecting here rather
than raising matters: an exception rolls back the claim transaction and blocks every
healthy package behind this one.
"""
reason = None
install_id = None
try:
payload = json.loads(payload_json)
except (TypeError, ValueError):
reason = "unreadable payload"
else:
# Valid JSON is not enough: a top-level array, string, number or
# null parses cleanly and then has no .get().
# Valid JSON is not enough: a top-level array/string/number parses cleanly.
if not isinstance(payload, dict):
reason = f"payload is {type(payload).__name__}, expected object"
else:
install_id = payload.get("install_id")
if not isinstance(install_id, str) or not install_id.strip():
reason = "payload has no usable install_id"
reason = (
None
if isinstance(install_id, str) and install_id.strip()
else "payload has no usable install_id"
)
if reason is not None:
logger.warning(
"Shared-metrics package %s cannot be sent (%s)", package_id, reason
)
logger.warning("Shared-metrics package %s cannot be sent (%s)", package_id, reason)
connection.execute(
"""
UPDATE package_outbox
@@ -407,16 +361,15 @@ class SharedMetricsSender:
# -- transmission ------------------------------------------------------
def _body(self, payload_json: str, transmitted_id: str) -> bytes:
@staticmethod
def _body(payload_json: str, transmitted_id: str) -> bytes:
"""Rebuild the exact bytes to send.
The payload is recomputed from the stored package rather than kept as a second copy:
json.dumps with these options is deterministic. The install_id is written from the frozen
``sent_install_id`` column rather than trusted implicitly, keeping "a resend is byte-
identical" anchored to one recorded value.
Recomputed from the stored package (json.dumps with these options is deterministic)
with install_id taken from the frozen ``sent_install_id`` column, keeping "a resend
is byte-identical" anchored to one recorded value.
"""
payload = json.loads(payload_json)
payload = dict(payload)
payload = dict(json.loads(payload_json))
payload["install_id"] = transmitted_id
return json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
@@ -430,54 +383,40 @@ class SharedMetricsSender:
) -> None:
"""Write send state for one package.
Guarded on send_state so a pass whose lease lapsed cannot resurrect a row another process
has already finished: without this, a slow sender could overwrite 'sent' back to 'pending'
and cause a re-send.
When ``token`` is given, the write is additionally compare-and-set on claim_token: it lands
only if THIS claim is still the current one. A claimant that lapsed and was superseded
writes zero rows — its settlement, backoff, and error strings all silently lose to the newer
claim's, which is the correct outcome.
Guarded on send_state so a pass whose lease lapsed cannot resurrect a row another
process already finished (overwriting 'sent' back to 'pending' would re-send). With
``token`` the write is also compare-and-set on claim_token: a superseded claimant
writes zero rows and its settlement/backoff/error silently lose to the newer claim.
"""
assignments = ", ".join(f"{name} = ?" for name in columns)
predicate = (
" AND (send_state IS NULL OR send_state = 'pending')"
if only_if_pending
else ""
)
predicate = " AND (send_state IS NULL OR send_state = 'pending')" if only_if_pending else ""
params: list = [*columns.values(), package_id]
if token is not None:
predicate += " AND claim_token = ?"
params.append(token)
with self._store._connection() as connection:
with write_txn(connection):
connection.execute(
f"UPDATE package_outbox SET {assignments} "
f"WHERE package_id = ?{predicate}",
params,
)
with self._write() as connection:
connection.execute(
f"UPDATE package_outbox SET {assignments} WHERE package_id = ?{predicate}",
params,
)
def _renew_claim(self, package_id: str, token: str | None) -> bool:
"""Atomically re-assert ownership and extend the lease. CAS, one row.
A read-only ownership check is not enough: a claimant whose lease expired while suspended
can pass the check (its token is still in the row if no one reclaimed yet) and then POST
while another process legitimately reclaims — the check-to-POST expiry race a seventh review
reproduced.
and only then pushing next_attempt_at a fresh lease into the future, so the upcoming POST
(30s timeout, well under the 300s lease) runs entirely inside renewed authority. rowcount ==
1 is the only grant.
A read-only ownership check is not enough: a claimant whose lease expired while
suspended still has its token in the row if nobody reclaimed yet, and would POST
while another process legitimately reclaims (check-to-POST expiry race). Pushing a
fresh lease in the same CAS keeps the upcoming POST (30s, well under the 300s lease)
entirely inside renewed authority. rowcount == 1 is the only grant.
"""
if token is None:
return False
try:
now = self._now()
lease_until = now + timedelta(seconds=_CLAIM_LEASE_SECONDS)
with self._store._connection() as connection:
with write_txn(connection):
cursor = connection.execute(
"""
with self._write() as connection:
cursor = connection.execute(
"""
UPDATE package_outbox
SET next_attempt_at = ?
WHERE package_id = ?
@@ -485,176 +424,120 @@ class SharedMetricsSender:
AND (send_state IS NULL OR send_state = 'pending')
AND next_attempt_at > ?
""",
(
_isoformat(lease_until),
package_id,
token,
_isoformat(now),
),
)
return cursor.rowcount == 1
(_isoformat(lease_until), package_id, token, _isoformat(now)),
)
return cursor.rowcount == 1
except Exception:
# If renewal itself fails, do not transmit on unproven authority.
logger.warning(
"Unable to renew shared-metrics claim", exc_info=True
)
logger.warning("Unable to renew shared-metrics claim", exc_info=True)
return False
def _defer(
self,
package_id: str,
delay_seconds: int,
reason: str,
*,
token: str | None = None,
self, package_id: str, delay_seconds: int, reason: str, *, token: str | None = None
) -> None:
# Defence in depth: no current caller can pass a non-positive delay
# (Retry-After is already clamped to [1, 86400] when parsed, and every
# other call site passes a positive constant), so this clamp is
# deliberately unreachable today and no test can distinguish it. It
# stays because a past deadline would make the row instantly
# re-eligible and let a pass spin on it — a cheap guard against a
# future caller that forgets.
delay = max(1, int(delay_seconds))
retry_at = self._now().timestamp() + delay
# Clamp to >= 1s: a past deadline would make the row instantly re-eligible and let
# a pass spin on it. Unreachable today (Retry-After is pre-clamped, other callers
# pass positive constants) — kept as a guard against a future caller.
retry_at = self._now().timestamp() + max(1, int(delay_seconds))
self._mark(
package_id,
token=token,
send_state="pending",
next_attempt_at=_isoformat(
datetime.fromtimestamp(retry_at, tz=timezone.utc)
),
next_attempt_at=_isoformat(datetime.fromtimestamp(retry_at, tz=timezone.utc)),
last_error=reason[:500],
)
def _send_one(self, package: dict) -> str:
"""Try one package. Returns 'sent', 'rejected', or 'deferred'.
Delivery is at-least-once. The pre-POST ownership check plus the token-fenced writes close
the claim->POST and settle-after-reclaim gaps, but a suspension landing MID-POST (bytes
already on the wire when the machine sleeps) can still duplicate: no client-side check can
revoke a request in flight.
Delivery is at-least-once. The pre-POST renewal plus token-fenced writes close the
claim->POST and settle-after-reclaim gaps, but a suspension landing MID-POST can
still duplicate: no client-side check can revoke a request in flight.
"""
package_id = package["package_id"]
token = package.get("claim_token")
body = self._body(package["payload_json"], package["derived"])
def defer(delay: int, reason: str) -> str:
self._defer(package_id, delay, reason, token=token)
return "deferred"
for attempt in range(1, self._max_attempts + 1):
# Atomically renew the claim before EVERY external POST. The
# renewal is compare-and-set on (token, pending, lease unexpired)
# and extends the lease past the request, so a suspended-then-
# resumed claimant whose lease lapsed yields here even if nobody
# has reclaimed yet — a read-only ownership check passed in that
# state and still double-sent (check-to-POST expiry race). The
# ingest key is minute-prefixed, so duplicates become distinct
# Renew before EVERY external POST (CAS on token, pending, lease unexpired), so a
# suspended-then-resumed claimant whose lease lapsed yields even before anyone
# reclaims. The ingest key is minute-prefixed, so duplicates become distinct
# stored objects, not overwrites.
if not self._renew_claim(package_id, token):
logger.info(
"Shared-metrics claim on %s superseded or expired; yielding",
package_id,
"Shared-metrics claim on %s superseded or expired; yielding", package_id
)
return "deferred"
try:
response = self._post(
self._endpoint, body, timeout=REQUEST_TIMEOUT_SECONDS
)
response = self._post(self._endpoint, body, timeout=REQUEST_TIMEOUT_SECONDS)
except Exception as exc: # transport failure: offline, DNS, TLS
reason = f"{type(exc).__name__}: {exc}"
if attempt >= self._max_attempts:
self._defer(
package_id, _FAILURE_BACKOFF_SECONDS, reason, token=token
else:
if response.status == 202:
self._mark(
package_id, token=token, send_state="sent",
sent_at=_isoformat(self._now()), last_error=None,
)
return "deferred"
self._sleep(self._backoff(attempt))
continue
if response.status == 202:
self._mark(
package_id,
token=token,
send_state="sent",
sent_at=_isoformat(self._now()),
last_error=None,
)
return "sent"
if response.status in _PERMANENT_STATUSES:
# Only statuses the contract (or the envelope schema) makes
# terminal. Everything else retries: 403 in particular is the
# ingest service's origin guard, which returns 403 during an
# edge/Transform-Rule misconfiguration — treating that as
# permanent would discard every package sent during the
# incident instead of retrying after recovery.
logger.warning(
"Telemetry package %s rejected with HTTP %s; not retrying",
package_id,
response.status,
)
self._mark(
package_id,
token=token,
send_state="rejected",
last_error=f"HTTP {response.status}: {response.body[:400]}",
)
return "rejected"
if response.status == 429:
self._defer(
package_id,
_retry_after_seconds(response.retry_after, _FAILURE_BACKOFF_SECONDS),
"rate limited",
token=token,
)
return "deferred"
# 5xx and anything unexpected: retryable.
reason = f"HTTP {response.status}"
return "sent"
if response.status in _PERMANENT_STATUSES:
# Only contract-terminal statuses. 403 in particular is the origin guard
# during an edge misconfiguration — treating it as permanent would
# discard every package sent during the incident.
logger.warning(
"Telemetry package %s rejected with HTTP %s; not retrying",
package_id,
response.status,
)
self._mark(
package_id, token=token, send_state="rejected",
last_error=f"HTTP {response.status}: {response.body[:400]}",
)
return "rejected"
if response.status == 429:
return defer(
_retry_after_seconds(response.retry_after, _FAILURE_BACKOFF_SECONDS),
"rate limited",
)
# 5xx and anything unexpected: retryable.
reason = f"HTTP {response.status}"
if attempt >= self._max_attempts:
self._defer(
package_id, _FAILURE_BACKOFF_SECONDS, reason, token=token
)
return "deferred"
return defer(_FAILURE_BACKOFF_SECONDS, reason)
self._sleep(self._backoff(attempt))
self._defer(
package_id, _FAILURE_BACKOFF_SECONDS, "attempts exhausted", token=token
)
return "deferred"
return defer(_FAILURE_BACKOFF_SECONDS, "attempts exhausted")
@staticmethod
def _backoff(attempt: int) -> float:
"""1s, 5s, 25s with full jitter."""
ceiling = _BACKOFF_BASE_SECONDS * (_BACKOFF_FACTOR ** (attempt - 1))
return random.uniform(0, ceiling)
return random.uniform(0, _BACKOFF_BASE_SECONDS * (_BACKOFF_FACTOR ** (attempt - 1)))
# -- entry point -------------------------------------------------------
def send_pending(self) -> SendOutcome:
"""Run one bounded pass. Never raises.
Claims and sends ONE package at a time so each row's lease only has to cover its own
transmission, and re-checks consent before every send so revoking `send` mid-pass stops the
remaining packages.
Claims and sends ONE package at a time so each row's lease only covers its own
transmission, and re-checks consent before every send so revoking `send` mid-pass
stops the remaining packages.
"""
outcome = SendOutcome()
seen: set[str] = set()
for _ in range(MAX_PACKAGES_PER_PASS):
if not self._still_consented():
# The user turned sending off while this pass was running.
# Stop without transmitting anything further, and reconcile
# so the window closes at its last confirmed moment. This is
# the same single writer every other observation point uses —
# not a separate recording mechanism.
# Stop transmitting and reconcile through the single consent writer so
# the window closes at its last confirmed moment.
logger.info("Shared-metrics sending disabled mid-pass; stopping")
self._reconcile(send_enabled=False)
break
try:
package = self._claim_next(self._now(), seen)
except Exception:
logger.warning(
"Unable to select shared-metrics packages", exc_info=True
)
logger.warning("Unable to select shared-metrics packages", exc_info=True)
break
if package is None:
break
@@ -662,42 +545,29 @@ class SharedMetricsSender:
seen.add(package["package_id"])
if package.get("skip"):
# Unusable row already marked rejected during the claim.
outcome.rejected += 1
continue
try:
result = self._send_one(package)
except Exception:
logger.warning("Unable to send shared-metrics package", exc_info=True)
outcome.deferred += 1
continue
if result == "sent":
outcome.sent += 1
elif result == "rejected":
outcome.rejected += 1
result = "rejected"
else:
outcome.deferred += 1
try:
result = self._send_one(package)
except Exception:
logger.warning("Unable to send shared-metrics package", exc_info=True)
result = "deferred"
setattr(outcome, result, getattr(outcome, result) + 1)
return outcome
def _reconcile(self, *, send_enabled: bool) -> None:
"""Run the single consent writer from within a pass."""
try:
with self._store._connection() as connection:
with write_txn(connection):
reconcile_send_consent(
connection, send_enabled, now=self._now()
)
with self._write() as connection:
reconcile_send_consent(connection, send_enabled, now=self._now())
except Exception:
logger.warning(
"Unable to reconcile shared-metrics consent", exc_info=True
)
logger.warning("Unable to reconcile shared-metrics consent", exc_info=True)
def _still_consented(self) -> bool:
"""Re-read profile-owned send consent.
Consent is a boundary, not cached config: docs promise ``send: false`` stops transmission
immediately, and a pass can run for minutes. Injected senders opt out via
consent_check=None.
Consent is a boundary, not cached config: docs promise ``send: false`` stops
transmission immediately, and a pass can run for minutes.
"""
if self._consent_check is None:
return True
@@ -705,8 +575,5 @@ class SharedMetricsSender:
return bool(self._consent_check())
except Exception:
# Fail CLOSED: if consent cannot be established, do not transmit.
logger.warning(
"Unable to confirm shared-metrics send consent; stopping",
exc_info=True,
)
logger.warning("Unable to confirm shared-metrics send consent; stopping", exc_info=True)
return False
hermes_cli/observability/shared_metrics_subscriber.py
+17 -19
View File
@@ -26,6 +26,20 @@ from .shared_metrics_contract import (
logger = logging.getLogger(__name__)
# Contract projections in match order; each yields (metric_name, dimensions) or None.
_COUNTERS = (
client_active_counter,
lambda event: _named(MODEL_ROUTE_METRIC, model_call_dimensions(event)),
lambda event: _named(TOOL_CALL_METRIC, tool_call_dimensions(event)),
task_counter,
tool_approval_counter,
skill_counter,
)
def _named(metric_name: str, dimensions: dict | None) -> tuple[str, dict] | None:
return None if dimensions is None else (metric_name, dimensions)
class SharedMetricsSubscriber:
"""Persist validated Hermes counters from Relay lifecycle events."""
@@ -56,17 +70,7 @@ class SharedMetricsSubscriber:
@staticmethod
def _classify(event: Any) -> tuple[str, dict] | None:
"""Return ``(metric_name, dimensions)`` for the first matching contract, else None."""
metric = client_active_counter(event)
if metric is not None:
return metric
for metric_name, project in (
(MODEL_ROUTE_METRIC, model_call_dimensions),
(TOOL_CALL_METRIC, tool_call_dimensions),
):
dimensions = project(event)
if dimensions is not None:
return metric_name, dimensions
return task_counter(event) or tool_approval_counter(event) or skill_counter(event)
return next((m for m in (project(event) for project in _COUNTERS) if m is not None), None)
def __call__(self, event: Any) -> None:
if self._runtime_id is not None:
@@ -87,14 +91,8 @@ class SharedMetricsSubscriber:
if metric_name == CLIENT_ACTIVE_METRIC:
self.store.record_client_active(self._client_resource)
else:
self.store.record_counter(
metric_name,
dimensions,
self._client_resource,
)
self.store.record_counter(metric_name, dimensions, self._client_resource)
except Exception:
logger.warning(
"Unable to persist the Hermes shared metric: %s",
metric_name,
exc_info=True,
"Unable to persist the Hermes shared metric: %s", metric_name, exc_info=True
)