feat: add explicit session metrics export
This commit is contained in:
Generated
+1
@@ -4097,6 +4097,7 @@ dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
"session-store",
|
||||
"tempfile",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
||||
@@ -8,3 +8,6 @@ license.workspace = true
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
serde_json = { workspace = true }
|
||||
session-store = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = { workspace = true }
|
||||
|
||||
@@ -1,29 +1,34 @@
|
||||
# session-metrics
|
||||
|
||||
## Role
|
||||
Session 単位の append-only な観測値を既存 session-log に記録し、明示的な
|
||||
metrics 読取 / JSONL export 経路で取り出すための小さなヘルパークレートです。
|
||||
|
||||
`session-metrics` records usage and memory/session metrics that are useful for diagnostics and maintenance.
|
||||
- 保存先は `session-store` の `LogEntry::Extension`
|
||||
- domain は `session.metrics`
|
||||
- metric は `name / ts / dimensions / value / correlation_id` の最小 envelope
|
||||
- `record_metric` / `record_metric_at` で現在の Segment に append する
|
||||
- `read_segment_metrics` は 1 Segment、`read_session_metrics` は Session 内の全
|
||||
Segment を読み、各 metric に `segment_id` と `compacted_from` を付ける
|
||||
- `export_metrics_jsonl` はその located metric を newline-delimited JSON にする
|
||||
- 通常の Session snapshot / Worker list / Worker detail は Extension を公開しない
|
||||
|
||||
## Boundaries
|
||||
compaction は `compact.start` を source Segment、`compact.finish` と
|
||||
`compact.post_request` を結果 Segment に記録する。同じ `correlation_id` と
|
||||
`SegmentStart.compacted_from` により、Segment をまたぐ attempt と次の通常 LLM
|
||||
request を結合できる。
|
||||
|
||||
Owns:
|
||||
```rust,ignore
|
||||
use session_metrics::{
|
||||
Metric, export_metrics_jsonl, read_session_metrics, record_metric_at,
|
||||
};
|
||||
|
||||
- metric record types and persistence helpers
|
||||
- explicit memory usage/read/reference observations where applicable
|
||||
- lightweight diagnostic data that should not become model context by itself
|
||||
let metric = Metric::now("compact.start")
|
||||
.with_value(12_345.0)
|
||||
.with_dimension("mode", "automatic")
|
||||
.with_correlation_id("018f6f8a-9822-7b11-8b35-706f30313700");
|
||||
record_metric_at(&store, location, &metric)?;
|
||||
|
||||
Does not own:
|
||||
|
||||
- prompt context packing (`agen`)
|
||||
- generated memory contents (`memory`)
|
||||
- provider billing semantics (`provider`)
|
||||
- UI status rendering (`tui`)
|
||||
|
||||
## Design notes
|
||||
|
||||
Metrics are observations. They may guide compaction, memory effectiveness analysis, or UX, but they are not authoritative conversation history and should not smuggle hidden state into model input.
|
||||
|
||||
## See also
|
||||
|
||||
- [`../../docs/design/memory-knowledge.md`](../../docs/design/memory-knowledge.md)
|
||||
- [`../../docs/design/compaction.md`](../../docs/design/compaction.md)
|
||||
let records = read_session_metrics(&store, location.session_id)?;
|
||||
let jsonl = export_metrics_jsonl(&records)?;
|
||||
# Ok::<(), Box<dyn std::error::Error>>(())
|
||||
```
|
||||
|
||||
@@ -12,13 +12,16 @@
|
||||
//! 観測できない値は `None` で明示する
|
||||
//! - 「後から埋まる値」(例: prune 発火直後の `cache_read_tokens`)は前 entry に
|
||||
//! 書き戻さず、`correlation_id` を共有する別 metric として流す。集計は読み手で join
|
||||
//! - 集計 / 可視化 API はこのクレートには無い。session-log を読めば取り出せる、
|
||||
//! までが到達点
|
||||
//! - 集計 / 可視化には [`read_session_metrics`] / [`read_segment_metrics`] /
|
||||
//! [`export_metrics_jsonl`] の明示的な metrics 専用経路を使う。通常の
|
||||
//! Session snapshot は `Extension` を公開しない。
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use session_store::{SegmentId, SessionId, Store, StoreError, save_extension, segment_log};
|
||||
use session_store::{
|
||||
LogEntry, SegmentId, SegmentOrigin, SessionId, Store, StoreError, save_extension, segment_log,
|
||||
};
|
||||
|
||||
/// Domain tag used in `LogEntry::Extension` for all metrics records.
|
||||
pub const DOMAIN: &str = "metrics";
|
||||
@@ -97,6 +100,165 @@ pub fn metrics_from_extensions(extensions: &[(String, serde_json::Value)]) -> Ve
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// A metric together with its durable Session/Segment origin.
|
||||
///
|
||||
/// `compacted_from` is copied from the Segment start record so readers can
|
||||
/// reconstruct compaction lineage without inferring relationships from metric
|
||||
/// names or timestamps.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
pub struct LocatedMetric {
|
||||
pub session_id: SessionId,
|
||||
pub segment_id: SegmentId,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub compacted_from: Option<SegmentOrigin>,
|
||||
pub log_index: usize,
|
||||
pub metric: Metric,
|
||||
}
|
||||
|
||||
#[derive(Debug)]
|
||||
pub enum SessionMetricsError {
|
||||
Store(StoreError),
|
||||
MissingSegmentStart {
|
||||
segment_id: SegmentId,
|
||||
},
|
||||
SessionMismatch {
|
||||
requested: SessionId,
|
||||
observed: SessionId,
|
||||
segment_id: SegmentId,
|
||||
},
|
||||
Encode(serde_json::Error),
|
||||
}
|
||||
|
||||
impl std::fmt::Display for SessionMetricsError {
|
||||
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
match self {
|
||||
Self::Store(error) => write!(formatter, "session metrics store error: {error}"),
|
||||
Self::MissingSegmentStart { segment_id } => {
|
||||
write!(formatter, "segment {segment_id} has no start record")
|
||||
}
|
||||
Self::SessionMismatch {
|
||||
requested,
|
||||
observed,
|
||||
segment_id,
|
||||
} => write!(
|
||||
formatter,
|
||||
"segment {segment_id} belongs to session {observed}, not {requested}"
|
||||
),
|
||||
Self::Encode(error) => write!(formatter, "session metrics encode error: {error}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl std::error::Error for SessionMetricsError {}
|
||||
|
||||
impl From<StoreError> for SessionMetricsError {
|
||||
fn from(error: StoreError) -> Self {
|
||||
Self::Store(error)
|
||||
}
|
||||
}
|
||||
|
||||
impl From<serde_json::Error> for SessionMetricsError {
|
||||
fn from(error: serde_json::Error) -> Self {
|
||||
Self::Encode(error)
|
||||
}
|
||||
}
|
||||
|
||||
/// Read metrics from one exact Segment.
|
||||
///
|
||||
/// This is an explicit metrics-only surface. It validates the Segment's
|
||||
/// durable start record and retains the log position of each metric.
|
||||
pub fn read_segment_metrics(
|
||||
store: &dyn Store,
|
||||
session_id: SessionId,
|
||||
segment_id: SegmentId,
|
||||
) -> Result<Vec<LocatedMetric>, SessionMetricsError> {
|
||||
let entries = store.read_all(session_id, segment_id)?;
|
||||
let (observed_session_id, compacted_from) = entries
|
||||
.iter()
|
||||
.find_map(|entry| match entry {
|
||||
LogEntry::AnnotatedSegmentStart {
|
||||
session_id,
|
||||
compacted_from,
|
||||
..
|
||||
} => Some((*session_id, compacted_from.clone())),
|
||||
_ => None,
|
||||
})
|
||||
.ok_or(SessionMetricsError::MissingSegmentStart { segment_id })?;
|
||||
if observed_session_id != session_id {
|
||||
return Err(SessionMetricsError::SessionMismatch {
|
||||
requested: session_id,
|
||||
observed: observed_session_id,
|
||||
segment_id,
|
||||
});
|
||||
}
|
||||
|
||||
Ok(entries
|
||||
.iter()
|
||||
.enumerate()
|
||||
.filter_map(|(log_index, entry)| match entry {
|
||||
LogEntry::Extension {
|
||||
domain, payload, ..
|
||||
} if domain == DOMAIN => {
|
||||
serde_json::from_value::<Metric>(payload.clone())
|
||||
.ok()
|
||||
.map(|metric| LocatedMetric {
|
||||
session_id,
|
||||
segment_id,
|
||||
compacted_from: compacted_from.clone(),
|
||||
log_index,
|
||||
metric,
|
||||
})
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Read every metric for a Session across all of its Segments.
|
||||
pub fn read_session_metrics(
|
||||
store: &dyn Store,
|
||||
session_id: SessionId,
|
||||
) -> Result<Vec<LocatedMetric>, SessionMetricsError> {
|
||||
let mut metrics = Vec::new();
|
||||
for segment_id in store.list_segments(session_id)? {
|
||||
metrics.extend(read_segment_metrics(store, session_id, segment_id)?);
|
||||
}
|
||||
metrics.sort_by(|left, right| {
|
||||
(
|
||||
left.metric.ts,
|
||||
metric_phase_order(&left.metric.name),
|
||||
left.segment_id,
|
||||
left.log_index,
|
||||
)
|
||||
.cmp(&(
|
||||
right.metric.ts,
|
||||
metric_phase_order(&right.metric.name),
|
||||
right.segment_id,
|
||||
right.log_index,
|
||||
))
|
||||
});
|
||||
Ok(metrics)
|
||||
}
|
||||
|
||||
/// Serialize located metrics as newline-delimited JSON for an explicit export.
|
||||
pub fn export_metrics_jsonl(metrics: &[LocatedMetric]) -> Result<String, SessionMetricsError> {
|
||||
let mut output = String::new();
|
||||
for metric in metrics {
|
||||
output.push_str(&serde_json::to_string(metric)?);
|
||||
output.push('\n');
|
||||
}
|
||||
Ok(output)
|
||||
}
|
||||
|
||||
fn metric_phase_order(name: &str) -> u8 {
|
||||
match name {
|
||||
"compact.start" => 0,
|
||||
"compact.finish" => 2,
|
||||
"compact.post_request" => 3,
|
||||
_ => 1,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
@@ -151,6 +313,101 @@ mod tests {
|
||||
assert_eq!(metrics[1].name, "b");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn explicit_reader_and_export_preserve_compaction_lineage() {
|
||||
use session_store::FsStore;
|
||||
|
||||
let temp = tempfile::tempdir().unwrap();
|
||||
let store = FsStore::new(temp.path()).unwrap();
|
||||
let session_id = SessionId::parse_str("018f6f8a-9822-7b11-8b35-706f30313701").unwrap();
|
||||
let source_segment_id =
|
||||
SegmentId::parse_str("018f6f8a-9822-7b11-8b35-706f30313702").unwrap();
|
||||
let result_segment_id =
|
||||
SegmentId::parse_str("018f6f8a-9822-7b11-8b35-706f30313703").unwrap();
|
||||
let correlation_id = "018f6f8a-9822-7b11-8b35-706f30313700";
|
||||
|
||||
store
|
||||
.create_segment(
|
||||
session_id,
|
||||
source_segment_id,
|
||||
&[LogEntry::AnnotatedSegmentStart {
|
||||
ts: 1,
|
||||
session_id,
|
||||
system_prompt: None,
|
||||
config: Default::default(),
|
||||
history: Vec::new(),
|
||||
forked_from: None,
|
||||
compacted_from: None,
|
||||
}],
|
||||
)
|
||||
.unwrap();
|
||||
let mut start = Metric::now("compact.start").with_correlation_id(correlation_id);
|
||||
start.ts = 10;
|
||||
record_metric(&store, session_id, source_segment_id, &start).unwrap();
|
||||
|
||||
let origin = SegmentOrigin {
|
||||
segment_id: source_segment_id,
|
||||
at_turn_index: 0,
|
||||
};
|
||||
store
|
||||
.create_segment(
|
||||
session_id,
|
||||
result_segment_id,
|
||||
&[LogEntry::AnnotatedSegmentStart {
|
||||
ts: 2,
|
||||
session_id,
|
||||
system_prompt: None,
|
||||
config: Default::default(),
|
||||
history: Vec::new(),
|
||||
forked_from: None,
|
||||
compacted_from: Some(origin.clone()),
|
||||
}],
|
||||
)
|
||||
.unwrap();
|
||||
let mut finish = Metric::now("compact.finish").with_correlation_id(correlation_id);
|
||||
finish.ts = 10;
|
||||
record_metric(&store, session_id, result_segment_id, &finish).unwrap();
|
||||
let mut post = Metric::now("compact.post_request").with_correlation_id(correlation_id);
|
||||
post.ts = 11;
|
||||
record_metric(&store, session_id, result_segment_id, &post).unwrap();
|
||||
|
||||
let source_metrics = read_segment_metrics(&store, session_id, source_segment_id).unwrap();
|
||||
assert_eq!(source_metrics.len(), 1);
|
||||
assert_eq!(source_metrics[0].compacted_from, None);
|
||||
|
||||
let metrics = read_session_metrics(&store, session_id).unwrap();
|
||||
assert_eq!(metrics.len(), 3);
|
||||
assert_eq!(metrics[0].metric.name, "compact.start");
|
||||
let finish = metrics
|
||||
.iter()
|
||||
.find(|record| record.metric.name == "compact.finish")
|
||||
.unwrap();
|
||||
assert_eq!(finish.segment_id, result_segment_id);
|
||||
assert_eq!(finish.compacted_from, Some(origin));
|
||||
assert!(
|
||||
metrics
|
||||
.iter()
|
||||
.all(|record| { record.metric.correlation_id.as_deref() == Some(correlation_id) })
|
||||
);
|
||||
|
||||
let exported = export_metrics_jsonl(&metrics).unwrap();
|
||||
let ordinary_snapshot = session_store::public_snapshot::project_current_session_snapshot(
|
||||
&store.read_all(session_id, result_segment_id).unwrap(),
|
||||
);
|
||||
let ordinary_json = serde_json::to_string(&ordinary_snapshot).unwrap();
|
||||
assert!(!ordinary_json.contains("compact.finish"));
|
||||
assert!(!ordinary_json.contains("compact.post_request"));
|
||||
let decoded = exported
|
||||
.lines()
|
||||
.map(|line| serde_json::from_str::<LocatedMetric>(line).unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(decoded, metrics);
|
||||
|
||||
let reopened = FsStore::new(temp.path()).unwrap();
|
||||
let restored = read_session_metrics(&reopened, session_id).unwrap();
|
||||
assert_eq!(restored, metrics);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn fold_skips_undeserializable_payloads() {
|
||||
// 将来 schema が変わって読めない payload も skip して落ちない。
|
||||
|
||||
Reference in New Issue
Block a user