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 a MatchResult that captures the matched template, extracted parameters, and similarity score.

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.

Public API

Types

Functions

Constants

Usage Examples

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()
stages := []string{"plan", "tool_call", "finish"}
coord, err := agentdrain.NewCoordinator(cfg, stages)
if err != nil {
	panic(err)
}
if err := coord.LoadDefaultWeights(); err != nil {
	panic(err)
}

evt := agentdrain.AgentEvent{
	Stage:  "tool_call",
	Fields: map[string]string{"tool": "bash", "status": "ok"},
}
result, report, err := coord.AnalyzeEvent(evt)
if err != nil {
	panic(err)
}
fmt.Println(result.Template, report.AnomalyScore)

flat := agentdrain.FlattenEvent(
	agentdrain.AgentEvent{
		Stage: "tool_call",
		Fields: map[string]string{
			"tool":       "search",
			"query":      "foo",
			"session_id": "abc123",
			"latency_ms": "42",
		},
	},
	[]string{"session_id"},
)
// flat == "stage=tool_call latency_ms=42 query=foo tool=search"

Design Decisions

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.

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.

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.

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.

Dependencies

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.

Thread Safety

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.

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