Skip to content

Commit bb9bd12

Browse files
github-actions[bot]github-actions[bot]
authored
Update package specifications for agentdrain and console (#53634)
1 parent 85fca34 commit bb9bd12

2 files changed

Lines changed: 210 additions & 1021 deletions

File tree

‎pkg/agentdrain/README.md‎

Lines changed: 73 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -4,69 +4,73 @@
44
55
## Overview
66

7-
The `agentdrain` package implements an online log-template miner inspired by the Drain algorithm and adapts it to `AgentEvent` records emitted by agentic workflow stages. It converts structured events into deterministic token streams, normalizes variable values with regex-based masking, groups similar events into clusters, and returns a `MatchResult` that captures the matched template, extracted parameters, and similarity score.
7+
The `agentdrain` package implements an online log-template miner inspired by the Drain algorithm and adapts it to `AgentEvent` records emitted by agentic workflow stages. It converts structured events into deterministic token streams, normalizes variable values with regex-based masking, groups similar events into clusters, and returns `MatchResult` values describing the matched template, extracted parameters, and similarity score.
88

9-
The package is designed for two related tasks: training on known-good runs and anomaly analysis of new runs. `Miner` handles a single stream of events, while `Coordinator` manages one `Miner` per stage so templates from `plan`, `tool_call`, `finish`, and other stages do not interfere with each other. Persisted snapshots and embedded default weights allow models to be reused across runs instead of starting from an empty state every time.
9+
The package supports two related workflows: training on known-good events and analyzing new events for anomalies. `Miner` manages a single stream of events, while `Coordinator` manages one `Miner` per stage so templates from `plan`, `tool_call`, `finish`, and other stages do not interfere with each other. Miner state can be serialized with `Snapshot`/`SnapshotCluster`, and coordinators can bootstrap from embedded default weights via `LoadDefaultWeights`.
10+
11+
The public API is intentionally small: event flattening and tokenization helpers, configurable masking, a concurrent miner, stage-aware coordination, and anomaly scoring. Internally, the package uses a parse tree and cluster store, but those remain unexported implementation details.
1012

1113
## Public API
1214

1315
### Types
1416

1517
| Type | Kind | Description |
1618
|------|------|-------------|
17-
| `AgentEvent` | struct | Structured event with a stage name and key/value fields to flatten, mask, and mine. |
18-
| `AnomalyDetector` | struct | Scores `MatchResult` values against similarity and rarity thresholds. |
19-
| `AnomalyReport` | struct | Summarizes anomaly flags, normalized score, and human-readable reason text. |
20-
| `Cluster` | struct | Template cluster with ID, tokenized template, observation count, and optional stage. |
21-
| `Config` | struct | Tuning parameters for masking, parse-tree depth, similarity threshold, and excluded fields. |
22-
| `Coordinator` | struct | Routes events to one `Miner` per stage and persists combined weights. |
19+
| `AgentEvent` | struct | Structured event with a `Stage` and key/value `Fields` used as miner input. |
20+
| `AnomalyDetector` | struct | Evaluates `MatchResult` values and produces `AnomalyReport` values using similarity and rarity thresholds. |
21+
| `AnomalyReport` | struct | Describes anomaly flags, normalized score, and human-readable reason text. |
22+
| `Cluster` | struct | Represents a mined template cluster with ID, tokenized template, size, and optional stage. |
23+
| `Config` | struct | Configures parse-tree depth, similarity threshold, wildcard token, masking rules, rarity threshold, and excluded fields. |
24+
| `Coordinator` | struct | Owns one `Miner` per stage and provides stage-aware training, analysis, and persistence. |
2325
| `MaskRule` | struct | Regex substitution rule applied before tokenization. |
24-
| `Masker` | struct | Compiled sequence of `MaskRule` values applied in order. |
25-
| `MatchResult` | struct | Result of matching or creating a cluster, including template, params, and similarity. |
26-
| `Miner` | struct | Concurrent single-stream Drain-style miner with training and analysis methods. |
27-
| `Snapshot` | struct | Serializable miner state used by `SaveJSON` and `LoadJSON`. |
28-
| `SnapshotCluster` | struct | Serializable form of a single `Cluster` within a `Snapshot`. |
26+
| `Masker` | struct | Compiled ordered set of `MaskRule` values that normalizes log lines. |
27+
| `MatchResult` | struct | Reports the cluster ID, rendered template, extracted params, similarity, and stage for a processed event. |
28+
| `Miner` | struct | Concurrent Drain-style miner for one event stream. |
29+
| `Snapshot` | struct | Serializable representation of a miner's config, clusters, and next cluster ID. |
30+
| `SnapshotCluster` | struct | Serializable representation of one cluster inside a `Snapshot`. |
2931

3032
### Functions
3133

3234
| Function | Signature | Description |
3335
|----------|-----------|-------------|
34-
| `(*Coordinator).AllClusters` | `func (c *Coordinator) AllClusters() map[string][]Cluster` | Returns a stage-to-cluster snapshot for every managed miner. |
35-
| `(*Coordinator).AnalyzeEvent` | `func (c *Coordinator) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)` | Routes an event to its stage's miner, analyzes it, then trains on it, returning both the match and the anomaly report. |
36-
| `(*Coordinator).LoadDefaultWeights` | `func (c *Coordinator) LoadDefaultWeights() error` | Loads the embedded default trained weights into all stage miners. |
37-
| `(*Coordinator).LoadSnapshots` | `func (c *Coordinator) LoadSnapshots(data map[string][]byte) error` | Restores stage miners from per-stage JSON snapshots, creating new miners for stages not in the original constructor input. |
38-
| `(*Coordinator).LoadWeightsJSON` | `func (c *Coordinator) LoadWeightsJSON(data []byte) error` | Restores all stage miners from a combined JSON blob produced by `SaveWeightsJSON`. |
39-
| `(*Coordinator).SaveSnapshots` | `func (c *Coordinator) SaveSnapshots() (map[string][]byte, error)` | Serializes each stage miner to per-stage JSON snapshots. |
40-
| `(*Coordinator).SaveWeightsJSON` | `func (c *Coordinator) SaveWeightsJSON() ([]byte, error)` | Serializes all stage snapshots into one combined JSON document. |
41-
| `(*Coordinator).TrainEvent` | `func (c *Coordinator) TrainEvent(evt AgentEvent) (*MatchResult, error)` | Routes an event to its stage's miner and trains on it, creating the stage miner on demand if needed. |
42-
| `(*AnomalyDetector).Analyze` | `func (d *AnomalyDetector) Analyze(result *MatchResult, isNew bool, cluster *Cluster) *AnomalyReport` | Produces an anomaly report for a match result and cluster context. |
43-
| `(*Masker).Mask` | `func (m *Masker) Mask(line string) string` | Applies all configured mask rules and returns the normalized line. |
44-
| `(*Miner).AnalyzeEvent` | `func (m *Miner) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)` | Flattens and analyzes an event without training, returning the would-be match and an anomaly report. |
45-
| `(*Miner).Clusters` | `func (m *Miner) Clusters() []Cluster` | Returns a safe snapshot of all known clusters in the miner. |
46-
| `(*Miner).LoadJSON` | `func (m *Miner) LoadJSON(data []byte) error` | Restores miner state (clusters and parse tree) from a JSON snapshot produced by `SaveJSON`. |
47-
| `(*Miner).SaveJSON` | `func (m *Miner) SaveJSON() ([]byte, error)` | Serializes the miner's clusters into a JSON `Snapshot`. |
48-
| `(*Miner).Train` | `func (m *Miner) Train(line string) (*MatchResult, error)` | Trains the miner on a raw line and returns the resulting match. |
49-
| `(*Miner).TrainEvent` | `func (m *Miner) TrainEvent(evt AgentEvent) (*MatchResult, error)` | Flattens an `AgentEvent` and trains the miner on the resulting line. |
50-
| `DefaultConfig` | `func DefaultConfig() Config` | Returns the production default miner configuration and default masking rules. |
51-
| `FlattenEvent` | `func FlattenEvent(evt AgentEvent, excludeFields []string) string` | Converts an event into deterministic `key=value` tokens with stage first and excluded fields omitted. |
36+
| `(*AnomalyDetector).Analyze` | `func (d *AnomalyDetector) Analyze(result *MatchResult, isNew bool, cluster *Cluster) *AnomalyReport` | Scores a match result and cluster context, producing anomaly flags, a normalized score, and reason text. |
37+
| `(*Coordinator).AllClusters` | `func (c *Coordinator) AllClusters() map[string][]Cluster` | Returns a snapshot of clusters for every registered stage. |
38+
| `(*Coordinator).AnalyzeEvent` | `func (c *Coordinator) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)` | Routes an event to its stage miner and returns both the match result and anomaly report. |
39+
| `(*Coordinator).LoadDefaultWeights` | `func (c *Coordinator) LoadDefaultWeights() error` | Loads embedded default weights from `data/default_weights.json` unless the embedded file is empty or `{}`. |
40+
| `(*Coordinator).LoadSnapshots` | `func (c *Coordinator) LoadSnapshots(snapshots map[string][]byte) error` | Restores per-stage miner snapshots, creating new stage miners when snapshots reference previously unknown stages. |
41+
| `(*Coordinator).LoadWeightsJSON` | `func (c *Coordinator) LoadWeightsJSON(data []byte) error` | Restores all stage miners from a combined JSON document produced by `SaveWeightsJSON`. |
42+
| `(*Coordinator).SaveSnapshots` | `func (c *Coordinator) SaveSnapshots() (map[string][]byte, error)` | Serializes each stage miner independently as JSON. |
43+
| `(*Coordinator).SaveWeightsJSON` | `func (c *Coordinator) SaveWeightsJSON() ([]byte, error)` | Serializes all stage snapshots into one combined JSON blob suitable for embedding as default weights. |
44+
| `(*Coordinator).TrainEvent` | `func (c *Coordinator) TrainEvent(evt AgentEvent) (*MatchResult, error)` | Routes an event to the miner for `evt.Stage` and updates that miner. |
45+
| `(*Masker).Mask` | `func (m *Masker) Mask(line string) string` | Applies all configured masking rules in order. |
46+
| `(*Miner).AnalyzeEvent` | `func (m *Miner) AnalyzeEvent(evt AgentEvent) (*MatchResult, *AnomalyReport, error)` | Performs inference, trains on the event, and returns both the resulting match and anomaly report. |
47+
| `(*Miner).Clusters` | `func (m *Miner) Clusters() []Cluster` | Returns a snapshot of all known clusters. |
48+
| `(*Miner).LoadJSON` | `func (m *Miner) LoadJSON(data []byte) error` | Replaces miner state from a JSON snapshot and rebuilds the parse tree. |
49+
| `(*Miner).SaveJSON` | `func (m *Miner) SaveJSON() ([]byte, error)` | Serializes miner state to JSON. |
50+
| `(*Miner).Train` | `func (m *Miner) Train(line string) (*MatchResult, error)` | Trains the miner on a raw line after masking and tokenization. |
51+
| `(*Miner).TrainEvent` | `func (m *Miner) TrainEvent(evt AgentEvent) (*MatchResult, error)` | Flattens an `AgentEvent`, trains on it, and propagates the event stage onto the result and cluster. |
52+
| `DefaultConfig` | `func DefaultConfig() Config` | Returns the built-in production defaults, including masking rules and excluded fields. |
53+
| `FlattenEvent` | `func FlattenEvent(evt AgentEvent, excludeFields []string) string` | Converts an event into a deterministic space-separated `key=value` token string with `stage=` first when present. |
5254
| `NewAnomalyDetector` | `func NewAnomalyDetector(simThreshold float64, rareClusterThreshold int) (*AnomalyDetector, error)` | Validates thresholds and constructs an anomaly detector. |
53-
| `NewCoordinator` | `func NewCoordinator(cfg Config, stages []string) (*Coordinator, error)` | Creates one stage-scoped miner for each supplied stage. |
54-
| `NewMasker` | `func NewMasker(rules []MaskRule) (*Masker, error)` | Compiles masking regexes into a reusable masker. |
55-
| `NewMiner` | `func NewMiner(cfg Config) (*Miner, error)` | Creates a miner with compiled mask rules, empty clusters, and a fresh parse tree. |
56-
| `StageSequence` | `func StageSequence(events []AgentEvent) string` | Returns a space-separated sequence of event stages. |
57-
| `Tokenize` | `func Tokenize(line string) []string` | Splits a masked line on whitespace boundaries. |
55+
| `NewCoordinator` | `func NewCoordinator(cfg Config, stages []string) (*Coordinator, error)` | Creates a stage-aware coordinator with one miner per supplied stage. |
56+
| `NewMasker` | `func NewMasker(rules []MaskRule) (*Masker, error)` | Compiles regex mask rules into a reusable masker. |
57+
| `NewMiner` | `func NewMiner(cfg Config) (*Miner, error)` | Constructs a miner with compiled mask rules, a fresh parse tree, and an empty cluster store. |
58+
| `StageSequence` | `func StageSequence(events []AgentEvent) string` | Returns the stages from a slice of events as a single space-separated string. |
59+
| `Tokenize` | `func Tokenize(line string) []string` | Splits a line on whitespace. |
5860

5961
### Constants
6062

6163
| Constant | Type | Value | Description |
6264
|----------|------|-------|-------------|
63-
| `AnomalyMaxScore` | untyped `float64` | `2.0` | Maximum raw anomaly score before normalization to `[0,1]`. |
64-
| `AnomalyWeightLow` | untyped `float64` | `0.7` | Weight applied when a known template matches below the configured similarity threshold. |
65-
| `AnomalyWeightNew` | untyped `float64` | `1.0` | Weight applied when analysis creates a brand-new cluster. |
66-
| `AnomalyWeightRare` | untyped `float64` | `0.3` | Weight applied when the matched cluster size is at or below the rare-cluster threshold. |
65+
| `AnomalyMaxScore` | untyped numeric constant | `2.0` | Maximum raw anomaly weight before normalization into the `[0,1]` score range. |
66+
| `AnomalyWeightLow` | untyped numeric constant | `0.7` | Weight added when a known cluster matches below the similarity threshold. |
67+
| `AnomalyWeightNew` | untyped numeric constant | `1.0` | Weight added when analysis creates a brand-new template cluster. |
68+
| `AnomalyWeightRare` | untyped numeric constant | `0.3` | Weight added when the matched cluster size is at or below the rare-cluster threshold. |
6769

6870
## Usage Examples
6971

72+
Examples below are taken from the package's spec tests and reflect the current public API.
73+
7074
```go
7175
cfg := agentdrain.DefaultConfig()
7276
miner, err := agentdrain.NewMiner(cfg)
@@ -87,59 +91,68 @@ fmt.Println(result.ClusterID)
8791

8892
```go
8993
cfg := agentdrain.DefaultConfig()
90-
stages := []string{"plan", "tool_call", "finish"}
91-
coord, err := agentdrain.NewCoordinator(cfg, stages)
94+
coord, err := agentdrain.NewCoordinator(cfg, []string{"plan", "tool_call", "finish"})
9295
if err != nil {
9396
panic(err)
9497
}
95-
if err := coord.LoadDefaultWeights(); err != nil {
96-
panic(err)
97-
}
9898

9999
evt := agentdrain.AgentEvent{
100-
Stage: "tool_call",
101-
Fields: map[string]string{"tool": "bash", "status": "ok"},
100+
Stage: "plan",
101+
Fields: map[string]string{"action": "evaluate", "step": "1"},
102102
}
103103
result, report, err := coord.AnalyzeEvent(evt)
104104
if err != nil {
105105
panic(err)
106106
}
107-
fmt.Println(result.Template, report.AnomalyScore)
107+
fmt.Println(result.Stage, report.AnomalyScore)
108108
```
109109

110110
```go
111111
flat := agentdrain.FlattenEvent(
112112
agentdrain.AgentEvent{
113113
Stage: "tool_call",
114114
Fields: map[string]string{
115-
"tool": "search",
116-
"query": "foo",
117-
"session_id": "abc123",
118-
"latency_ms": "42",
115+
"session_id": "abc-123",
116+
"action": "start",
119117
},
120118
},
121119
[]string{"session_id"},
122120
)
123-
// flat == "stage=tool_call latency_ms=42 query=foo tool=search"
121+
fmt.Println(flat)
122+
// Output: stage=tool_call action=start
124123
```
125124

126-
## Design Decisions
125+
```go
126+
masker, err := agentdrain.NewMasker([]agentdrain.MaskRule{{
127+
Name: "number_test",
128+
Pattern: `\d+`,
129+
Replacement: "<NUM>",
130+
}})
131+
if err != nil {
132+
panic(err)
133+
}
134+
fmt.Println(masker.Mask("step 42 completed"))
135+
```
127136

128-
`FlattenEvent` MUST emit deterministic output: the `stage=` token is first when present, remaining keys are sorted alphabetically, and excluded fields are omitted. This keeps clustering stable across map iteration order and allows persisted weights to be reused reliably.
137+
## Design Decisions
129138

130-
`AnalyzeEvent` performs inference before updating training state, then trains on the same event and scores the result against the matched or created cluster. New-template anomalies and low-similarity anomalies are intentionally mutually exclusive: a brand-new cluster is already anomalous without also being labeled low similarity.
139+
`FlattenEvent` is deterministic by design: it emits `stage=` first when present, sorts remaining field keys alphabetically, and omits explicitly excluded fields. This makes clustering stable across Go map iteration order and allows saved weights to remain reusable.
131140

132-
`Coordinator` SHOULD be used when events belong to semantically different stages. Each stage receives its own miner so templates from unrelated phases do not merge into the same cluster space. `LoadSnapshots` MAY create new stage miners when snapshots contain stages that were not part of the original constructor input.
141+
`Miner.AnalyzeEvent` performs inference before training, then trains on the same event and scores the resulting cluster with `AnomalyDetector`. The anomaly flags intentionally treat “new template” and “low similarity” as mutually exclusive so a brand-new cluster is not double-counted as both conditions.
133142

134-
The package embeds default trained weights in `data/default_weights.json`. Callers MAY use `LoadDefaultWeights` to start from a pre-trained baseline instead of training from scratch.
143+
`Coordinator` isolates miners by stage. This prevents templates from unrelated workflow phases from merging into the same cluster space and supports persistence as either per-stage snapshots or one combined weights document. The embedded default-weights mechanism provides an opt-in pre-trained baseline without exposing embedding details through additional API surface.
135144

136145
## Dependencies
137146

138-
Internal dependencies include `pkg/logger` for debug logging, `pkg/setutil` for exclusion-set membership, and `pkg/sliceutil` for slice and map helpers. External dependencies are limited to the Go standard library.
147+
Internal package dependencies include `pkg/logger` for debug logging, `pkg/setutil` for exclusion-set membership checks, and `pkg/sliceutil` for collection helpers used while flattening events. The package also embeds `data/default_weights.json` for coordinator bootstrapping.
148+
149+
External dependencies for production code are limited to the Go standard library, including `encoding/json`, `regexp`, `sort`, `strings`, `sync`, and `embed` support.
139150

140151
## Thread Safety
141152

142-
`Miner` and `Coordinator` are safe for concurrent use. `Miner` protects mutable state with an internal `sync.RWMutex`; training, analysis, and load paths acquire write locks, while cluster snapshots and persistence reads acquire read locks. `Coordinator` protects its stage-to-miner map with its own `sync.RWMutex` and delegates per-stage concurrency to each `Miner`.
153+
`Miner` is safe for concurrent use. It protects mutable state with an internal `sync.RWMutex`; training and load operations take the write lock, while cluster snapshots and JSON save operations take the read lock.
154+
155+
`Coordinator` is also safe for concurrent use. It protects its stage-to-miner map with its own `sync.RWMutex` and relies on each contained `Miner` for per-stage concurrency control.
143156

144157
---
145158

0 commit comments

Comments
 (0)