Agentic observability

import atexit
import json
import logging
import os
import queue
from datetime import datetime, timezone
from logging.handlers import QueueHandler, QueueListener


class HourlyJsonHandler(logging.Handler):
    """Appends each record as one JSON line to <root>/<service>/<YYYY-MM-DD>/<HH>/<host>.ndjson, in UTC."""

    def __init__(self, root, service, host):
        super().__init__()
        self.root, self.service, self.host = root, service, host
        self.path, self.file = None, None

    def emit(self, record):
        ts = datetime.fromtimestamp(record.created, timezone.utc)
        path = f"{self.root}/{self.service}/{ts:%Y-%m-%d/%H}/{self.host}.ndjson"
        if path != self.path:
            if self.file:
                self.file.close()
            os.makedirs(os.path.dirname(path), exist_ok=True)
            self.file, self.path = open(path, "a"), path
        entry = {
            "ts": ts.isoformat(timespec="milliseconds"),
            "level": record.levelname,
            "service": self.service,
            "host": self.host,
            "msg": record.getMessage(),
            **getattr(record, "fields", {}),
        }
        self.file.write(json.dumps(entry) + "\n")
        self.file.flush()


def archil_logger(service, root="/mnt/logs", host=None):
    """A logger whose records are written to the disk by a background thread, so logging never blocks on storage."""
    records = queue.Queue(-1)
    handler = HourlyJsonHandler(root, service, host or os.environ.get("INSTANCE") or os.uname().nodename)
    listener = QueueListener(records, handler)
    listener.start()
    atexit.register(listener.stop)  # write out queued records on exit
    logger = logging.getLogger(service)
    logger.addHandler(QueueHandler(records))
    logger.setLevel(logging.INFO)
    return logger

import os
import random
import time
import uuid
from logging import ERROR, INFO

from archil_logging import archil_logger

BREAK_FLAG = "/tmp/break-checkout"
HEIC_FLAG = "/tmp/heic-uploads"
api, checkout, worker = (archil_logger(name) for name in ["api", "checkout", "worker"])


def log(logger, level, msg, **fields):
    logger.log(level, msg, extra={"fields": fields})


def pay(request_id):
    order_id = f"ord_{uuid.uuid4().hex[:10]}"
    amount = random.choice([1299, 2499, 4999, 8900])
    if os.path.exists(BREAK_FLAG) and random.random() < 0.35:
        ms = random.randint(5000, 5100)
        log(checkout, ERROR, "payment authorization failed", request_id=request_id, order_id=order_id,
            amount_cents=amount, provider="payments-gateway", duration_ms=ms,
            error="upstream timeout after 5000ms (pool=payments-gateway, in_use=64/64)")
        return 502, ms
    ms = round(random.lognormvariate(5.0, 0.3))
    log(checkout, INFO, "payment authorized", request_id=request_id, order_id=order_id,
        amount_cents=amount, provider="payments-gateway", duration_ms=ms)
    return 200, ms


while True:
    path = random.choices(["/", "/search", "/cart", "/checkout"], weights=[4, 3, 2, 1])[0]
    request_id = f"req_{uuid.uuid4().hex[:12]}"
    if path == "/checkout":
        status, ms = pay(request_id)
        ms += random.randint(5, 20)
    else:
        status, ms = 200, round(random.lognormvariate(3, 0.5))
    log(api, ERROR if status >= 500 else INFO, "request", method="POST" if path == "/checkout" else "GET",
        path=path, status=status, duration_ms=ms, request_id=request_id, user_id=f"u_{random.randint(1, 5000)}")

    if random.random() < 0.05:
        job = random.choice(["send_receipt", "sync_inventory", "resize_image"])
        if job == "send_receipt" and random.random() < 0.08:
            log(worker, ERROR, "job failed", job=job, attempt=1, error="smtp 550: mailbox unavailable")
        elif job == "resize_image" and os.path.exists(HEIC_FLAG) and random.random() < 0.5:
            log(worker, ERROR, "job failed", job=job, attempt=1, error="unsupported image format: heic (thumbnail skipped)")
        else:
            log(worker, INFO, "job done", job=job, duration_ms=round(random.lognormvariate(4, 0.7)))

    time.sleep(random.uniform(0.01, 0.05))

// Usage: node vl2svg.mjs < chart.vl.json > chart.svg
import { readFileSync } from "node:fs";
import * as vega from "vega";
import { compile } from "vega-lite";

const spec = JSON.parse(readFileSync(0, "utf8"));
const view = new vega.View(vega.parse(compile(spec).spec), { renderer: "none" });
process.stdout.write(await view.toSVG());

import * as archil from "disk";
import { readFile } from "node:fs/promises";

const { disk: logs } = await archil.createDisk({ name: "app-logs" });
const { disk: agent } = await archil.createDisk({ name: "obs-agent" });

// The application host. Its services write their logs straight to the disk.
const sandbox = await archil.createSandbox({
  name: "shop",
  baseImage: "python:3.13",
  vcpuCount: 1,
  memSizeMiB: 1024,
  idleTtlSeconds: 0,
  mounts: [{ disk: logs, path: "/mnt/logs", conditional: true }],
});
for (const file of ["shop.py", "archil_logging.py"]) {
  await sandbox.files.uploadFile(await readFile(file), `/root/${file}`);
}
await sandbox.exec("archil-sandbox services create shop --env INSTANCE=shop-1 -- python3 /root/shop.py");

// The agent's workspace: DuckDB for SQL, and Vega to render charts as SVG.
await agent.putObject("bin/vl2svg.mjs", await readFile("vl2svg.mjs"));
const res = await archil.exec({
  disks: { agent: { disk: agent, conditional: true } },
  command: [
    "cd agent",
    "curl -sfL -o /tmp/duckdb.zip https://github.com/duckdb/duckdb/releases/download/v1.5.6/duckdb_cli-linux-amd64.zip",
    "unzip -oq /tmp/duckdb.zip -d bin",
    "npm install --no-audit --no-fund --silent vega@6.4.0 vega-lite@6.5.0",
    "bin/duckdb --version",
  ].join(" && "),
});
if (res.exitCode !== 0) throw new Error(res.stderr);
console.log(res.stdout);

import * as archil from "disk";

const [logs] = await archil.listDisks({ name: "app-logs" });
const [agent] = await archil.listDisks({ name: "obs-agent" });

const { stdout, timing } = await archil.exec({
  disks: { logs: { disk: logs, readOnly: true }, agent: { disk: agent, readOnly: true } },
  command: `agent/bin/duckdb -c "
    SELECT service, level, count(*) AS lines, max(ts) AS latest
    FROM read_json('logs/*/*/*/*.ndjson', union_by_name=true)
    GROUP BY ALL ORDER BY ALL"`,
});
console.log(stdout, `${timing.totalMs} ms`);

import * as archil from "disk";
import { createDiskTools } from "@archildata/ai-sdk";

export const [logs] = await archil.listDisks({ name: "app-logs" });
export const [agentDisk] = await archil.listDisks({ name: "obs-agent" });

// Logs are mounted read-only: the agent can read every line but can't change one.
export const tools = createDiskTools(
  archil.workspace({
    logs: { disk: logs, readOnly: true },
    agent: { disk: agentDisk, conditional: true },
  }),
);

export const model = "anthropic/claude-sonnet-5.5";

// Each agent step resends the conversation so far. Caching bills the repeated part at a fraction of the input price.
export const providerOptions = { gateway: { caching: "auto" as const } };

export const instructions = `
You are the on-call engineer for a web shop with three services: api, checkout, and worker.

Logs are under /logs/<service>/<YYYY-MM-DD>/<HH>/<host>.ndjson, partitioned by UTC hour. Each line is one JSON object
with ts (UTC), level, service, msg, host, and service-specific fields. request_id links an api request to the
checkout call it made. /logs is read-only. Your own files live under /agent.

Query logs with SQL in run_bash, from the workspace root:
  agent/bin/duckdb -c "SELECT ... FROM read_json('logs/*/2026-11-06/*/*.ndjson', union_by_name=true)"
Glob only the days and hours you need. Use the grep tool to find specific lines.
Base every claim on a query or a log line you have read, and quote request_ids or timestamps as evidence.
`.trim();

-- Compare the last 15 minutes with the hour before it. Returns one row per anomaly, or nothing.
-- triage.ts fills in the list of log files for the current and previous two hours.
SET VARIABLE t1 = (now() AT TIME ZONE 'UTC')::TIMESTAMP;
SET VARIABLE t0 = getvariable('t1') - INTERVAL 15 MINUTE;
SET VARIABLE b0 = getvariable('t0') - INTERVAL 60 MINUTE;

CREATE TEMP TABLE events AS
SELECT ts, service, path, duration_ms, ts >= getvariable('t0') AS recent,
       level = 'ERROR' OR coalesce(status, 0) >= 500 AS is_error,
       -- An error's fingerprint is its message with numbers masked, so repeats group together.
       msg || coalesce(': ' || regexp_replace(error, '[0-9]+', 'N', 'g'), '') AS fingerprint,
       coalesce(error, msg) AS example
FROM read_json([FILES], union_by_name=true)
WHERE ts >= getvariable('b0') AND ts < getvariable('t1');

WITH volume AS (
  SELECT service, count(*) FILTER (recent) AS recent_n, count(*) FILTER (NOT recent) AS baseline_n
  FROM events GROUP BY service
),
errors AS (
  SELECT service, fingerprint, any_value(example) AS example, min(ts) FILTER (recent) AS first_seen,
         count(*) FILTER (recent) AS recent_errors, count(*) FILTER (NOT recent) AS baseline_errors
  FROM events WHERE is_error GROUP BY ALL
),
latency AS (
  SELECT service, path,
         quantile_cont(duration_ms, 0.95) FILTER (recent) AS recent_p95,
         quantile_cont(duration_ms, 0.95) FILTER (NOT recent) AS baseline_p95,
         count(*) FILTER (recent) AS recent_n, count(*) FILTER (NOT recent) AS baseline_n
  FROM events WHERE duration_ms IS NOT NULL GROUP BY ALL
)
-- Errors that are new, or whose share of the service's lines at least doubled
SELECT 'errors' AS kind, service, fingerprint AS signal, example, first_seen,
       round(100 * recent_errors / recent_n, 2) AS recent_pct,
       round(100 * baseline_errors / nullif(baseline_n, 0), 2) AS baseline_pct,
       recent_errors AS recent_count, baseline_errors AS baseline_count
FROM errors JOIN volume USING (service)
WHERE recent_errors >= 5
  AND (baseline_errors = 0 OR recent_errors / recent_n >= 2 * baseline_errors / baseline_n)
UNION ALL
-- p95 latency that at least doubled on a path with enough traffic to compare
SELECT 'latency', service, coalesce(path, '(all)'), NULL, NULL,
       round(recent_p95), round(baseline_p95), recent_n, baseline_n
FROM latency
WHERE recent_n >= 20 AND baseline_n >= 20 AND recent_p95 >= 2 * baseline_p95
ORDER BY kind, recent_count DESC;

import { experimental_decide as decide, Output, ToolLoopAgent, isStepCount } from "ai";
import * as archil from "disk";
import { readFile } from "node:fs/promises";
import { z } from "zod";
import { agentDisk, instructions, logs, model, providerOptions, tools } from "./agent";

const cost = (metadata: any) => Number(metadata?.gateway?.cost ?? 0);

// 1. Detect: deterministic SQL next to the data. No model.
const sql = await readFile("detect.sql", "utf8");
const gate = await archil.exec({
  disks: { logs: { disk: logs, readOnly: true }, agent: { disk: agentDisk, readOnly: true } },
  command: `files=$(for h in 0 1 2; do ls logs/*/$(date -u -d "-$h hour" +%F/%H)/*.ndjson 2>/dev/null; done | sed "s/.*/'&'/" | paste -sd, -)
[ -n "$files" ] || exit 0
sed "s|\\[FILES\\]|[$files]|" <<'SQL' | agent/bin/duckdb -json
${sql}
SQL`,
});
if (gate.exitCode !== 0) throw new Error(gate.stderr);
const anomalies: Record<string, string | number | null>[] = gate.stdout.trim() ? JSON.parse(gate.stdout) : [];
console.log(`detect: ${anomalies.length} anomalies in ${gate.timing.totalMs} ms`);
if (anomalies.length === 0) process.exit(0);

// 2. Classify: a decision model decides which anomalies deserve an investigation.
const known = await agentDisk.getObject("triage/known-issues.md").then((b) => new TextDecoder().decode(b), () => "");
const decision = await decide({
  model: "typesafe-ai/jev",
  state: [
    { type: "text", text: `Issues already reported, with their rates:\n${known || "(none)"}` },
    { type: "json", value: anomalies },
  ],
  questions: Object.fromEntries(
    anomalies.map((anomaly, i) => [
      `anomaly_${i}`,
      {
        type: "choice",
        instructions: `Does this anomaly from a web shop's logs need an engineer to investigate? ${JSON.stringify(anomaly)}`,
        criteria: {
          investigate: "Customers are likely affected (failed requests, payments, or orders, or much slower responses), and it is new or clearly worse than an issue already reported.",
          ignore: "Little or no customer impact, such as a skipped background task or a rejected email, or an issue already reported at a similar rate.",
        },
      } as const,
    ]),
  ),
});
const flagged = anomalies.filter((anomaly, i) => {
  const answer = decision.answers[`anomaly_${i}`];
  const p = answer.probabilities?.[answer.choice];
  console.log(`classify: ${answer.choice} (${p}) ${anomaly.kind} ${anomaly.service} ${anomaly.signal}`);
  return answer.choice === "investigate";
});
console.log(`classify: $${cost(decision.providerMetadata).toFixed(6)}`);
if (flagged.length === 0) process.exit(0);

// 3. Investigate: a frontier model digs into what the first two steps flagged.
const Issue = z.object({
  severity: z.enum(["critical", "warning"]),
  service: z.string(),
  title: z.string().describe("One line a human can act on, e.g. 'Checkout payments timing out'"),
  started: z.string().describe("UTC timestamp of the first occurrence"),
  impact: z.string().describe("Quantified impact, e.g. '31% of checkouts failing (412 of 1,320)'"),
  evidence: z.array(z.string()).describe("Two or three log lines or query results that prove it"),
  nextStep: z.string(),
});

const agent = new ToolLoopAgent({
  model,
  instructions,
  tools,
  providerOptions,
  stopWhen: isStepCount(40),
  output: Output.object({ schema: z.object({ issues: z.array(Issue), summary: z.string() }) }),
});

const now = new Date().toISOString();
const result = await agent.generate({
  prompt: `
It is ${now}. An anomaly detector compared the last 15 minutes of logs with the hour before and flagged:
${JSON.stringify(flagged, null, 2)}

Investigate each one: confirm it, find when it started, quantify how many requests, jobs, or users it affects, and check
whether other services are involved. Trace a few request_ids across services. Anomalies with the same cause are one issue.
Report only these anomalies. A classifier has already judged the detector's other findings to be low-impact.
Write a short Markdown report to /agent/triage/reports/${now.slice(0, 16).replace(":", "")}.md, and add each issue to
/agent/triage/known-issues.md with its current rate.`,
});

const { issues, summary } = result.output;
const usage = result.totalUsage;
const spent = result.steps.reduce((sum, step) => sum + cost(step.providerMetadata), 0);
console.log(
  `investigate: ${result.steps.length} steps, ${usage.inputTokens} input tokens ` +
    `(${usage.inputTokenDetails.cacheReadTokens ?? 0} cached), $${spent.toFixed(4)}\n`,
);
console.log(summary);
for (const issue of issues) {
  console.log(`\n[${issue.severity}] ${issue.service}: ${issue.title}`);
  console.log(`  started ${issue.started}; ${issue.impact}`);
  for (const line of issue.evidence) console.log(`  - ${line}`);
  console.log(`  next: ${issue.nextStep}`);
}

if (issues.length > 0 && process.env.SLACK_WEBHOOK_URL) {
  const text = issues.map((i) => `*[${i.severity}] ${i.service}: ${i.title}*\n${i.impact}\n${i.nextStep}`);
  await fetch(process.env.SLACK_WEBHOOK_URL, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ text: text.join("\n\n") }),
  });
}

import * as archil from "disk";

const sandbox = (await archil.listSandboxes()).find((s) => s.name === "shop")!;
// Break checkout, and at the same time start a new, harmless error: users uploading HEIC images.
await sandbox.exec("touch /tmp/break-checkout /tmp/heic-uploads");
console.log(`checkout broken at ${new Date().toISOString()}`);

import { Output, ToolLoopAgent, isStepCount } from "ai";
import { z } from "zod";
import { agentDisk, instructions, model, providerOptions, tools } from "./agent";

const question = process.argv.slice(2).join(" ");

const agent = new ToolLoopAgent({
  model,
  instructions,
  tools,
  providerOptions,
  stopWhen: isStepCount(40),
  output: Output.object({
    schema: z.object({
      slug: z.string().describe("Short kebab-case name of the dashboard directory"),
      findings: z.string().describe("Two or three sentences answering the question"),
    }),
  }),
});

const { output, steps } = await agent.generate({
  prompt: `
Build a dashboard that answers: "${question}"

Explore the logs first, then write /agent/dashboards/<slug>/build.sh. When it runs from the workspace root, it must
query the logs with agent/bin/duckdb -json, render 2 to 4 Vega-Lite charts to SVG with
node agent/bin/vl2svg.mjs < spec.json > chart.svg, and write /agent/dashboards/<slug>/index.html: one static page with
a title, a one-line takeaway above each chart, and the SVGs inlined. The page is served with scripts disabled, so don't
use JavaScript. Give charts a fixed width of about 800 pixels and keep each one under 500 data points.
Use time windows relative to now, so the script still works when it runs again later.
Run build.sh, then read index.html back to check it.`,
});

const share = await agentDisk.share(`dashboards/${output.slug}/index.html`, { expiresIn: 7 * 24 * 3600 });
const spent = steps.reduce((sum, step) => sum + Number((step.providerMetadata?.gateway as any)?.cost ?? 0), 0);
console.log(`${steps.length} steps, $${spent.toFixed(4)}\n`);
console.log(output.findings);
console.log(`\n${share.url}`);

import * as archil from "disk";

const slug = process.argv[2];
const [logs] = await archil.listDisks({ name: "app-logs" });
const [agent] = await archil.listDisks({ name: "obs-agent" });

const res = await archil.exec({
  disks: {
    logs: { disk: logs, readOnly: true },
    agent: { disk: agent, conditional: true },
  },
  command: `bash agent/dashboards/${slug}/build.sh`,
});
if (res.exitCode !== 0) throw new Error(res.stderr);
console.log(`rebuilt in ${res.timing.totalMs} ms`);
console.log((await agent.share(`dashboards/${slug}/index.html`)).url);

import * as archil from "disk";

const sandbox = (await archil.listSandboxes()).find((s) => s.name === "shop")!;
await sandbox.stop();
await sandbox.delete();
for (const name of ["app-logs", "obs-agent"]) {
  const [disk] = await archil.listDisks({ name });
  await disk.delete();
}