Error Aggregation

Error Aggregation Skill

Overview

The error aggregation skill groups raw harness error events by domain and frequency so you receive a structured, deduplicated view — not a wall of log lines. It handles:

• Deduplication — identical stack traces and similar messages are merged into a single ErrorGroup.
• Trend detection — each group carries a trend of rising, falling, or stable based on how event rates compare across the analysis window.
• Domain breakdown — groups are indexed by the logical component that produced them (e.g. gate_runner, lsp, deploy).
• JSON serialisation — the view is pre-serialised into compact JSON for embedding in agent context budgets.

Workflow

Do you want a one-shot CLI summary? → Run the CLI

Do you want to embed the view inside an agent? → Programmatic usage

Do you want to let Claude query the view interactively? → Agent query runner

CLI Usage

# Summarise the most critical errors from the last 60 minutes:
python -m harness_skills.error_query_agent \
    --log-file /var/log/harness/errors.ndjson \
    --window 60 \
    --prompt "Which domains are producing the most errors right now?"

# Print the raw aggregation JSON and exit (no Claude call):
python -m harness_skills.error_query_agent \
    --log-file /var/log/harness/errors.ndjson \
    --json-summary

# Restrict to a custom window and model:
python -m harness_skills.error_query_agent \
    --log-file /var/log/harness/errors.ndjson \
    --window 30 \
    --model claude-opus-4-6 \
    --prompt "Are there any rising errors in the deploy domain?"

The helper script scripts/query_errors.py wraps the CLI for quick interactive use without installing the full package.

Programmatic Usage

1 — Load from an NDJSON log file

from harness_skills.error_aggregation import load_errors_from_log, aggregate_errors, errors_to_json_summary

records = load_errors_from_log("/var/log/harness/errors.ndjson", window_minutes=60)
view    = aggregate_errors(records, window_minutes=60)

# Compact JSON for agent context:
print(errors_to_json_summary(view, top_n=20))

2 — Build from in-memory records

from datetime import datetime, timezone
from harness_skills.error_aggregation import ErrorRecord, aggregate_errors

records = [
    ErrorRecord(
        timestamp=datetime.now(tz=timezone.utc),
        domain="gate_runner",
        error_type="TimeoutError",
        message="Gate timed out after 30 s",
        severity="error",
    ),
    # … more records …
]
view = aggregate_errors(records, window_minutes=60)

3 — Query helpers

from harness_skills.error_aggregation import top_errors, errors_by_domain, domain_summary

# Top 10 errors globally:
top_errors(view, n=10)

# Top 5 errors in a specific domain:
top_errors(view, n=5, domain="gate_runner")

# {domain -> [ErrorGroup]} index:
errors_by_domain(view)

# Bird's-eye domain overview (sorted by total error count):
domain_summary(view)
# → [{"domain": "gate_runner", "total_errors": 42, "distinct_patterns": 3,
#     "top_severity": "error", "rising_patterns": 1}, …]

Agent Query Runner

The run_error_query coroutine wires the aggregation view into a ClaudeSDKClient session with two MCP tools:

import asyncio
from harness_skills.error_query_agent import run_error_query
from harness_skills.error_aggregation import load_errors_from_log, aggregate_errors

records = load_errors_from_log("/var/log/harness/errors.ndjson")
view    = aggregate_errors(records)

result = asyncio.run(
    run_error_query(
        prompt="Which domain has the most rising errors in the last hour?",
        view=view,
        model="claude-opus-4-6",
        max_turns=6,
    )
)
print(result)

Or use build_error_tools to get the MCP server object and plug it into an existing agent session:

from harness_skills.error_query_agent import build_error_tools
from claude_agent_sdk import ClaudeAgentOptions

server  = build_error_tools(records=records)
options = ClaudeAgentOptions(
    mcp_servers={"error-aggregation": server},
    allowed_tools=["query_recent_errors", "get_error_domain_list"],
    # … other options …
)

NDJSON Log Format

load_errors_from_log reads newline-delimited JSON files. Required fields per line:

{"timestamp": "2026-03-14T10:00:00Z", "domain": "gate_runner", "error_type": "TimeoutError", "message": "Gate timed out after 30 s"}

Optional fields: stack_hint (first stack-trace line, improves deduplication), severity ("error" | "warning" | "critical", default "error").

Lines that cannot be parsed are silently skipped.

Data Structures

ErrorRecord

ErrorGroup

ErrorAggregationView

Key Files