[agent]
name = "log-analyzer"
version = "0.1.1"
description = "Analyzes log files - sweeps many files or time windows in parallel, then identifies anomalies, trends, and error patterns via a scripted analyze⇄script loop, maintaining a severity-ranked findings index"
entry_stage = "ingest"
[tool_permissions]
read_file = "allow"
list_dir = "allow"
bash = "ask"
write_file = "ask"
# ─── Stage 1: Ingest ──────────────────────────────────────────────────────────
[stages.ingest]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Read log files, identify format and structure"
available_tools = ["read_file", "list_dir", "bash", "context_write"]
max_iterations = 15
system_prompt = """
Ingest the log files named in `task`. Determine:
1. Log format (JSON, syslog, Apache, custom, etc.)
2. Time range covered and total volume
3. Available fields/columns
4. Obvious structural issues (corrupted lines, encoding problems)
Use bash to sample rather than reading whole huge files - the first N lines, a
line count, a file type probe. Reach for whatever your shell actually provides
(`head`/`wc`/`file` on Unix, `Get-Content -TotalCount` and `Measure-Object` on
Windows). Write a short format/structure summary to `findings` (context_write) so
the analyze stage starts oriented, and name every log file you found in it with
its size and the time range it covers - that inventory is what the next stage
divides the sweep over. Raw sampled content lands in `logs`.
"""
[stages.ingest.tool_routing]
default_region = "conversation"
[stages.ingest.tool_routing.overrides]
read_file = "logs"
bash = "logs"
[stages.ingest.transitions.split_logs]
hint = "Log format identified, ready to sweep the files"
transform = "direct"
# Un-exhaustible escape (lint: dead-end-possible): when every looping
# target has spent its max_revisits budget, the run can still move forward
# to the deliverable instead of dead-ending.
[stages.ingest.transitions.report]
hint = "Analysis is out of revisits - report what ingestion established"
condition = "dead_end"
transform = "compact"
[stages.ingest.transitions.error_recovery]
condition = "error"
transform = "direct"
# ─── Stage 2: Split logs ──────────────────────────────────────────────────────
# One worker per log file or time window, so a hundred rotated files get swept at
# once instead of one after another - and each worker reads its own slice into a
# context nothing else is competing for. The workers hand back findings; the
# analyze stage takes them from there and digs into what they turned up.
[stages.split_logs]
mode = "fan_out"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Split the logs into files or time windows, one sweeper each"
available_tools = []
max_iterations = 4
# split_logs ↔ error_recovery is a cycle, so it needs a ceiling. One retry is
# enough to route around a transient failure; a second means the logs cannot be
# divided usefully and the run should analyze what it has rather than keep
# splitting.
max_revisits = 1
worker_stage = "scan_worker"
merge_stage = "analyze"
max_workers = 4
# A file that will not parse is a gap in the sweep, not a reason to throw the
# other workers' findings away. The failure is named in the results and the
# analyze stage decides what to do about it.
on_worker_failure = "continue"
# Where the workers' findings land, and the budget their shares divide. A region
# of its own rather than the conversation, which is carrying the message history
# alongside them.
results_region = "worker_findings"
# A ceiling on slices, not just on concurrency. Eight is enough breadth for any
# log set - a rotation with more files than that wants time windows, not one
# worker each - and past it every worker's share of the region is too small to
# carry the counts that make a finding evidence.
max_items = 8
split_prompt = """
Split the sweep into slices that can be read independently, one work item each.
A slice is one log file, one service's files, or one time window of a single
large file - whatever divides the data without two workers counting the same
lines. Use the file inventory `findings` holds.
Output ONLY a JSON array, starting with '[' and ending with ']'. No prose, no
markdown fences. Each item:
[{"id": "<short-slug>", "context": {"slice": "<the file path(s), and the time window if the file is split by time>", "format": "<the log format and its fields, verbatim from findings>", "looking_for": "<what this run is trying to find, from the task>"}}]
Every item MUST carry the format and what the run is looking for: workers do not
share this agent's context, so the work item is all they get. A worker that does
not know the field layout can only grep for the word "error".
One log file is ONE item. Split a single file by time only when it is too large
to sweep in one pass, and give each window an explicit start and end.
"""
[stages.split_logs.transitions.error_recovery]
condition = "error"
transform = "direct"
# ─── Worker: sweep one slice ─────────────────────────────────────────────────
# Entered as a sub-agent, one per slice. It hands back findings, which the
# analyze stage reads. `require_output` is what makes that a guarantee rather
# than a hope. bash is the point of the worker: counting a million lines is a
# shell job, and its whole output stays in the worker's own context rather than
# this one's.
[stages.scan_worker]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Sweep one log file or time window for errors, anomalies, and trends"
available_tools = ["read_file", "bash", "submit_output"]
allow_as_worker = true
require_output = true
max_iterations = 20
system_prompt = """
Sweep ONE slice of the logs. Your work item names the file (and time window),
the format its lines are in, and what this run is looking for - it is all the
context you get, so work from it.
Count rather than skim. Aggregate with whatever your shell runs (awk/grep/sed or
python via bash on Unix, PowerShell or python on Windows): error counts by type,
rates per interval, status-code and latency distributions, the first and last
occurrence of anything unusual. Read the surrounding lines of a spike before
calling it one.
Submit what you found with submit_output. Every finding carries its evidence -
counts and timestamps from your own commands, never an impression. An hour you
could not parse is itself a finding; say which lines defeated you.
A quiet slice is a result: say so, with the volume you swept, rather than
inflating routine noise into a finding. Only your submitted findings reach the
merge - anything else you write is lost.
"""
[stages.scan_worker.output]
format = "markdown"
instructions = "One finding per line, as `<severity> | <finding> | <evidence: counts/timestamps>` with severity in {critical, warning, info}, most severe first. Start with a `info | swept <slice> | <line count>, <time range>` line so the merge knows what was covered."
[stages.scan_worker.transitions]
# ─── Stage 3: Analyze ─────────────────────────────────────────────────────────
[stages.analyze]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-opus-5" }, { provider = "openai", model = "gpt-5.5" }, { provider = "google", model = "gemini-3.1-pro-preview" }, { provider = "openrouter", model = "deepseek/deepseek-v4-pro" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Identify anomalies, trends, and error patterns"
available_tools = ["read_file", "bash", "context_write", "context_append"]
max_iterations = 25
max_revisits = 3
transition_prompt = """
Decide what happens next:
- If you need a script to parse/filter/aggregate the data more precisely, respond with: script
- If you have enough findings to produce a report, respond with: report
"""
system_prompt = """
The sweepers' findings are in the `worker_findings` region, one block per slice.
Start there: fold each finding into `findings`, merge the ones that are the same
error seen in two files, and treat a worker reported as FAILED as a slice nobody
swept - say so, or sweep it here. What no single worker could see is yours: the
same failure crossing files, and the order events happened in across them.
Then analyze the log data for:
1. Error patterns - recurring errors, error spikes, cascading failures
2. Anomalies - unusual request rates, latency outliers, unexpected status codes
3. Trends - gradual degradation, growing queue depths, memory creep
4. Correlations - errors that co-occur, time-based patterns
5. Security signals - auth failures, unusual access, injection attempts
Quantify everything (timestamps, counts, rates). For each finding, append to
`findings` (context_append): `<SEVERITY> | <finding> | <evidence: counts/timestamps>`
with SEVERITY in {critical, warning, info}. Keep a running tally in `severity_index`
(context_write), e.g. "critical: 2, warning: 5, info: 3". Use the `script` stage
when you need precise parsing/aggregation you can't eyeball.
If `error_report` says a previous stage hit its iteration cap, that stage was cut
off before finishing - treat its output as incomplete and re-check what it missed.
"""
[stages.analyze.tool_routing]
default_region = "conversation"
[stages.analyze.tool_routing.overrides]
read_file = "logs"
bash = "scratch"
[stages.analyze.transitions.script]
hint = "Need a script to parse/filter/aggregate log data"
transform = "compact"
[stages.analyze.transitions.report]
hint = "Analysis complete - ready to produce report"
transform = "compact"
[stages.analyze.transitions.error_recovery]
condition = "error"
transform = "direct"
# ─── Stage 4: Script ──────────────────────────────────────────────────────────
[stages.script]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Write and execute scripts to parse, filter, and aggregate log data"
available_tools = ["read_file", "bash"]
max_iterations = 20
max_revisits = 5
transition_prompt = """
Your script has produced results. Decide what happens next:
- If you need to refine this script or write another, respond with: script
- If the results answer the analysis questions, respond with: analyze
"""
system_prompt = """
Write scripts to process the logs, in whatever your shell can run (awk/grep/sed
or python via bash on Unix, PowerShell or python on Windows):
- Parse structured fields, filter by time/severity/pattern
- Aggregate counts, rates, percentiles
- Detect anomalies via statistical thresholds
- Correlate events across sources
Write small, focused scripts; test each before building on it. Script code and
output land in `scripts`. Iterate until the output answers the analysis questions,
then return to analyze.
"""
[stages.script.tool_routing]
default_region = "scripts"
[stages.script.tool_routing.overrides]
read_file = "logs"
[stages.script.transitions.script]
hint = "Refine or write another script"
transform = "direct"
[stages.script.transitions.analyze]
hint = "Script results ready - return to analysis"
transform = "compact"
# Un-exhaustible escape (lint: dead-end-possible): when every looping
# target has spent its max_revisits budget, the run can still move forward
# to the deliverable instead of dead-ending.
[stages.script.transitions.report]
hint = "Analysis is out of revisits - report the findings so far"
condition = "dead_end"
transform = "compact"
[stages.script.transitions.error_recovery]
condition = "error"
transform = "direct"
# ─── Stage 5: Report ──────────────────────────────────────────────────────────
[stages.report]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-opus-5" }, { provider = "openai", model = "gpt-5.5" }, { provider = "google", model = "gemini-3.1-pro-preview" }, { provider = "openrouter", model = "deepseek/deepseek-v4-pro" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Produce a structured analysis report with findings"
available_tools = ["write_file", "read_file"]
max_iterations = 10
# Generating the final report - don't inject mid-run user messages into it.
accepts_messages = false
system_prompt = """
Write a structured log analysis report (write_file) from the `findings` region,
structured by the `severity_index` tally. Include:
- Executive summary (top findings in 3-5 bullets)
- Log source overview (files, format, time range, volume)
- Findings by severity (🔴 critical / 🟡 warning / 🔵 info)
- For each: description, evidence (counts/timestamps), impact
- Trends and patterns observed
- Recommendations (what to fix, what to monitor)
- Appendix: key log excerpts and script outputs
Be precise - timestamps, counts, percentages, specific log lines. The report's
severity counts must match `severity_index`.
"""
[stages.report.transitions.summary]
hint = "The report is written"
# ─── Stage 6: Error recovery ──────────────────────────────────────────────────
[stages.error_recovery]
mode = "autonomous"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Recover from a failed read or script execution"
available_tools = ["read_file", "bash"]
max_iterations = 10
max_revisits = 1
system_prompt = """
A read or script execution failed. The `error_report` region holds the error
text the runtime captured - read it first, diagnose it (a malformed log line, a
script bug, a missing file), note a workaround, then return to analyze to
continue with the data available.
"""
# Un-exhaustible escape (lint: dead-end-possible): when every looping
# target has spent its max_revisits budget, the run can still move forward
# to the deliverable instead of dead-ending.
[stages.error_recovery.transitions.report]
hint = "Recovery is not converging - report the findings so far"
transform = "compact"
[stages.error_recovery.transitions.analyze]
hint = "Recovered - resume analysis"
transform = "compact"
[compaction]
provider = "anthropic"
model = "claude-sonnet-5"
# ─── Final output ────────────────────────────────────────────────────────────
# Terminal. `mode = "output"` grants `submit_output`, requires the call, and
# lets the run end here. No file tools on purpose: this stage reports, it does
# not work.
[stages.summary]
mode = "output"
model = { models = [{ provider = "anthropic", model = "claude-sonnet-5" }, { provider = "openai", model = "gpt-5.4-mini" }, { provider = "google", model = "gemini-3.5-flash" }, { provider = "openrouter", model = "deepseek/deepseek-v4-flash" }, { provider = "ollama", model = "qwen3.5:9b" }] }
description = "Say what the logs showed"
max_iterations = 8
system_prompt = """
Say what these logs showed, for whoever asked for it.
Lead with the diagnosis: what went wrong, and how confident you are. Then the
evidence that points at it, and anything you could not explain. Name the report
file you wrote.
If you found nothing conclusive, say that plainly rather than dressing up a
guess.
"""
[stages.summary.transitions]
# ─── Context layout ───────────────────────────────────────────────────────────
[context.regions]
task = { kind = "pinned", budget = "2%", max_tokens = 3000, required = true, seed = "task", required_message = "Name the log file(s)/dir and what to look for via --task." }
severity_index = { kind = "pinned", budget = "2%", max_tokens = 2000 }
findings = { kind = "sliding_window", max_items = 30, budget = "12%", max_tokens = 15000, strategy = "bulk", overflow = 10 }
# The sweepers' submitted findings, on the way into the analyze stage. Its budget
# is what each worker's share is divided from, so it matches the findings region
# it feeds. Clearable: once analyze has folded the findings in, the raw worker
# report is safe to drop.
worker_findings = { kind = "clearable", budget = "12%", max_tokens = 15000 }
logs = { kind = "temporary", budget = "30%", max_tokens = 50000 }
scripts = { kind = "compacting", budget = "20%", compact_at = "80%", threshold_tokens = 20000, max_tokens = 30000 }
scripts_history = { kind = "compact_history", source_region = "scripts", budget = "3%", max_tokens = 10000 }
conversation = { kind = "sliding_window", max_items = 30, budget = "12%", max_tokens = 15000, strategy = "bulk", overflow = 10 }
scratch = { kind = "clearable", budget = "6%", max_tokens = 6000 }
# Written by the RUNTIME on an abnormal stage ending (issue #154): a failed
# inference call's error text, or a note that a stage hit its iteration cap.
# Pinned so it survives the edge transform into the stage that acts on it.
error_report = { kind = "pinned", budget = "1%", max_tokens = 2000 }