agentdrain

agentdrain Package

Drain-style log template mining and anomaly scoring for structured agent pipeline events.

Overview

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.

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.

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.

Public API

Types

Functions

Constants

Usage Examples

Examples below are taken from the package's spec tests and reflect the current public API.

cfg := agentdrain.DefaultConfig()
miner, err := agentdrain.NewMiner(cfg)
if err != nil {
	panic(err)
}

evt := agentdrain.AgentEvent{
	Stage:  "plan",
	Fields: map[string]string{"action": "start", "step": "1"},
}
result, err := miner.TrainEvent(evt)
if err != nil {
	panic(err)
}
fmt.Println(result.ClusterID)

cfg := agentdrain.DefaultConfig()
coord, err := agentdrain.NewCoordinator(cfg, []string{"plan", "tool_call", "finish"})
if err != nil {
	panic(err)
}

evt := agentdrain.AgentEvent{
	Stage:  "plan",
	Fields: map[string]string{"action": "evaluate", "step": "1"},
}
result, report, err := coord.AnalyzeEvent(evt)
if err != nil {
	panic(err)
}
fmt.Println(result.Stage, report.AnomalyScore)

flat := agentdrain.FlattenEvent(
	agentdrain.AgentEvent{
		Stage: "tool_call",
		Fields: map[string]string{
			"session_id": "abc-123",
			"action":     "start",
		},
	},
	[]string{"session_id"},
)
fmt.Println(flat)
// Output: stage=tool_call action=start

masker, err := agentdrain.NewMasker([]agentdrain.MaskRule{{
	Name:        "number_test",
	Pattern:     `\d+`,
	Replacement: "<NUM>",
}})
if err != nil {
	panic(err)
}
fmt.Println(masker.Mask("step 42 completed"))

Design Decisions

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.

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.

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.

Dependencies

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.

External dependencies for production code are limited to the Go standard library, including encoding/json, regexp, sort, strings, sync, and embed support.

Thread Safety

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.

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.

This specification is automatically maintained by the spec-extractor workflow.