Build a durable AI agent

npm create gnl@latest oncall-agent
cd oncall-agent
pnpm install

// gnl.config.ts — written by `gnl init`, not by hand.
import { PostgresStorage } from '@gnldev/durable/postgres';

export default {
  // Postgres, because you said the journal lives in one. `DATABASE_URL` is read at startup and
  // is NOT defaulted — a journal that silently falls back to a local file is a journal you
  // discover is empty in production.
  storage: new PostgresStorage({ connectionString: process.env.DATABASE_URL! }),
  agents: { assistant, 'charge-demo': { ...chargeDemo, tools: { chargeOrder } } },
  preset: 'assistant',
};

// src/models/nvidia.ts — written by `gnl add model nvidia`.
import { createOpenAICompatible } from '@ai-sdk/openai-compatible';

const nim = createOpenAICompatible({
  name: 'nvidia',
  baseURL: 'https://integrate.api.nvidia.com/v1',
  apiKey: process.env.NVIDIA_API_KEY ?? '',
});

export const model = nim(process.env.NVIDIA_MODEL ?? 'poolside/laguna-xs-2.1');

// then in src/agents/assistant.ts, two lines:
//   import { model } from '../models/nvidia.js';
//   …and in the config object:  model,   (replacing echoModel())

// src/tools.ts
import { tool } from 'ai';
import { z } from 'zod';

/** Stands in for the fleet, so the example can count what actually happened. */
export const fleet = { restarts: [] as string[] };

/**
 * The one thing this agent can DO — and the reason the framework is here.
 *
 * Restarting a service twice is two outages. Unlike a double charge there is nothing to refund,
 * which is why "it probably won't happen twice" is not a design.
 */
export const restartService = tool({
  description: 'Restarts a service. Disruptive.',
  inputSchema: z.object({ service: z.string() }),
  execute: async ({ service }) => {
    fleet.restarts.push(service);
    return { service, restarted: true };
  },
});

/**
 * A read — and the source of every secret in this tutorial.
 *
 * An agent that can act on a service is an agent that reads that service's logs, and real log lines
 * carry connection strings and customer addresses.
 */
export const readLog = tool({
  description: 'Reads the tail of a service log.',
  inputSchema: z.object({ service: z.string() }),
  execute: async ({ service }) => ({
    service,
    lines: [
      'INFO  boot: connected postgres://svc:hunter2@db-primary/orders',
      'INFO  request: customer [email protected] placed order',
      'ERROR gc: heap 1.9GB / 2.0GB',
    ],
  }),
});

// Reading a log twice is harmless, so this opts OUT of the default side-effect treatment.
// Part 7 measures exactly what that changes. Never put it on a tool that writes.
Object.assign(readLog, { idempotent: true });

export const tools = { restartService, readLog };

// src/agent.ts — the config, and nothing else.
//
// Note what this file does NOT do: it never calls the agent. It describes one. Who calls it is the
// next file's problem, and for HTTP traffic the answer is "the framework does".
import type { CreateGnlConfig } from '@gnldev/durable';
import { storage } from './storage.js';
import { MODEL } from './model.js';
import { tools } from './tools.js';

export const config: CreateGnlConfig = {
  storage,
  agents: {
    ops: {
      model: MODEL,
      tools,
      // The system prompt is the agent's JOB DESCRIPTION — the one place you say what it is for.
      // Everything you add from here constrains what it MAY DO. They are not substitutes: a prompt
      // asking the model nicely not to restart things is not a guard.
      system:
        'You are an on-call engineer. Read the service log before acting, and never restart a ' +
        'service unless the log shows it is genuinely stuck.',
      // A budget, not a target: caps how many tool-call rounds one run may take, which is what stops
      // a confused model looping.
      maxSteps: 4,
    },
  },
};

// src/server.ts — mount it. This is the whole integration.
import { Hono } from 'hono';
import { serve } from '@hono/node-server';
import { createRestApi } from '@gnldev/server';
import { config } from './agent.js';

const app = new Hono();
app.mount('/api', createRestApi(config, { title: 'On-call', allowOpenAccess: true }));

serve({ fetch: app.fetch, port: 3000, hostname: '127.0.0.1' });

// src/agent.ts — add the guard
//
// A guard runs before EVERY tool call and answers allow | require-approval. This one stops exactly
// one tool: a gate on a read costs an engineer's attention and buys nothing, and a system that asks
// about everything is a system people learn to click through.
export const guard = ({ toolName, args }: any) =>
  toolName === 'restartService'
    ? { action: 'require-approval' as const, reason: `restarting ${args?.service} interrupts live traffic` }
    : { action: 'allow' as const };

export const gnl = createGnl({
  storage,
  agents: { ops: { model: MODEL, tools, guard, system: '…', maxSteps: 4 } },
});

/** Continues a suspended run once a human has answered. */
export const resume = (runId: string, approvals: Record<string, boolean>) =>
  resumeRun(runId, { journal: storage.runs, model: MODEL, tools, guard, stopWhen: stepCountIs(4), approvals });

const r = await gnl.run('ops', { runId: `triage:${incidentId}`, prompt });

if (r.interrupts.length) {
  const it = r.interrupts[0];
  console.log(`⏸  ${it.toolName} — ${it.reason}`);

  // In a real app this is an HTTP endpoint a human hits (Part 19), not a line below the run.
  // The point of suspending is that the answer does not have to come from this process, or this hour.
  const done = await resume(`triage:${incidentId}`, { [it.toolCallId]: true });
}

// src/agent.ts — what the model is allowed to SEE
import { piiRedactor } from '@gnldev/processors';

export const gnl = createGnl({
  storage,
  processors: [
    piiRedactor({
      // Off by default because it changes what gets written to the journal. For an agent whose job
      // is reading logs, it is the entire point.
      redactToolResults: true,
      // The built-in set is PII — email, phone, card, IBAN, SSN, IP. A database password is a
      // CREDENTIAL, not PII, and is measurably not covered. This closes that, and keeps the HOST
      // readable: at 3am "which database" is the question, and [REDACTED] answers none of it.
      extraPatterns: [
        { name: 'dsn', pattern: /(?<=:\/\/)[^\s:@/]+:[^\s@/]+(?=@)/g, mask: '[REDACTED_DSN]' },
      ],
    }),
  ],
  agents: { ops: { model: MODEL, tools, guard, system: '…', maxSteps: 4 } },
});

// src/agent.ts
import { defaultRules } from '@gnldev/tool-schema';

export const gnl = createGnl({
  storage,
  processors: [...],
  // Rewrites each tool's JSON Schema for whichever provider is actually underneath. Invisible until
  // the day you point this same config at a provider in strict mode and a schema shape it silently
  // rejects takes the agent down.
  schemaCompat: defaultRules,
  agents: { ops: { model: MODEL, tools, guard, system: '…', maxSteps: 4 } },
});

// src/embed.ts — one function from string to vector, used by memory AND retrieval.
const EMBED_MODEL = process.env.NVIDIA_EMBED_MODEL ?? 'nvidia/nemotron-3-embed-1b';

/**
 * NVIDIA requires `input_type`: 'query' for what you search WITH, 'passage' for what you search
 * THROUGH. Sending the wrong one still returns a vector, so nothing errors — retrieval just gets
 * quietly worse, which is the hardest kind of bug to notice.
 */
async function embedWith(input: string, inputType: 'query' | 'passage'): Promise<number[]> {
  const res = await fetch('https://integrate.api.nvidia.com/v1/embeddings', {
    method: 'POST',
    headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.NVIDIA_API_KEY}` },
    body: JSON.stringify({ model: EMBED_MODEL, input: [input], input_type: inputType }),
  });
  if (!res.ok) throw new Error(`embeddings ${res.status}`);
  const json = await res.json();
  return json.data[0].embedding;
}

export const embed = (t: string) => embedWith(t, 'query');
export const embedPassage = (t: string) => embedWith(t, 'passage');

// src/agent.ts — memory
import { AgentMemory } from '@gnldev/memory';
import { z } from 'zod';
import { embed } from './embed.js';

const memory = new AgentMemory({
  storage,               // same Postgres — messages, threads and vectors all land there
  embed,
  recentN: 8,            // the last 8 turns verbatim
  // Recall scoped to the SERVICE, not the caller: at 3am what matters is what happened to `checkout`
  // last time, whoever was holding the pager then.
  recall: { topK: 3, scope: 'resource' },
  // Structured facts the agent keeps across turns, validated by a schema rather than hoped for.
  workingMemory: { schema: z.object({ service: z.string().optional(), severity: z.string().optional() }) },
});

export const gnl = createGnl({
  storage,
  memory,                // TOP-LEVEL, not inside the agent — an agent-level `memory` is ignored
  processors: [...],
  schemaCompat: defaultRules,
  agents: { ops: { model: MODEL, tools, guard, system: '…', maxSteps: 4 } },
});

await gnl.run('ops', {
  runId: `triage:${incidentId}`,
  prompt,
  threadId: incidentId,   // the conversation — memory continuity across turns
  resourceId: service,    // WHOSE memory — recall is scoped to this
});

// src/runbooks.ts — what the agent reads before it acts.
//
// An on-call agent that improvises is worse than no agent. It has to answer "what does THIS
// organisation do about this alert", and that lives in prose someone wrote at 3am last time.
import { InMemoryVectorStore, indexDocuments, createRagTool } from '@gnldev/rag';
import { createCache } from '@gnldev/cache';
import { storage } from './storage.js';
import { embed, embedPassage } from './embed.js';

const RUNBOOKS = [
  { id: 'rb-memory', text:
    'RUNBOOK — memory saturation. Applies when memory_rss_mb exceeds 1800. Read the service log ' +
    'first; a leak shows as rising RSS with flat request volume. If it matches, restart the ' +
    'service — a restart is disruptive and needs SRE approval. Then watch for 30 minutes: a leak ' +
    'that returns inside that window is a code bug, escalate rather than restart again.' },
  { id: 'rb-latency', text:
    'RUNBOOK — latency regression. Applies when p99 exceeds 2000ms. Do NOT restart. Latency is ' +
    'almost always downstream: check the database, then the cache hit rate.' },
  { id: 'rb-escalation', text:
    'RUNBOOK — escalation. Page the on-call engineer once and only once per incident. A second page ' +
    'for the same incident trains people to ignore the first.' },
];

// The corpus is re-indexed on boot. Cache the embeddings and you pay for each text once, ever —
// across runs, across restarts, because the key is the content, not the run.
const cache = createCache(storage.cache, 'embed');
const cachedPassage = (t: string) => cache.getOrCompute(t, () => embedPassage(t));

export async function buildRunbookTool() {
  const store = new InMemoryVectorStore();
  await indexDocuments(store, cachedPassage, RUNBOOKS);
  // Returns an ORDINARY tool — the agent calls it by name like any other.
  return createRagTool({ store, embed, topK: 1,
    description: 'Finds the runbook for an alert. Always read this before acting.' });
}

// src/workflow.ts — procedure, not more prompt.
//
// The agent's job is JUDGEMENT: which runbook applies, what the numbers mean. The pipeline's job is
// PROCEDURE: triage, escalate, record. Procedure written into a system prompt is procedure you
// cannot audit and cannot replay — and "page once per incident" is procedure.
import { workflow, step } from '@gnldev/workflow';

// A step's output is the next step's input. Each is journaled under `runId#stepId`.
const triage = step('triage', async (alert: Alert, ctx) => {
  // HERE is where you call the agent yourself — you are driving it from your own code, not serving
  // an HTTP request. This is the case `gnl.run()` exists for.
  const r = await gnl.run('ops', {
    runId: `triage:${alert.incidentId}`,
    prompt: alert.text,
    threadId: alert.incidentId,
    resourceId: alert.service,
  });
  const it = r.interrupts?.[0];
  return { ...alert, diagnosis: r.text ?? '', suspended: !!it, awaiting: it };
});

const escalate = step('escalate', async (t) => {
  if (t.severity !== 'sev1') return { ...t, paged: false };
  await page('sre-oncall', t.incidentId);
  return { ...t, paged: true };
});

const close = step('close', async (r) => { await record(r); return r; });

export const incident = workflow<Alert>().then(triage).then(escalate).then(close);

// src/agent.ts — register it alongside the agents.
export const config: CreateGnlConfig = {
  storage,
  agents: { ops: { … } },
  workflows: { incident },     // now reachable at POST /api/workflows/incident/run
};

// src/ops.ts — how alerts arrive, and what watches afterwards.
import { enqueue, createWorker } from '@gnldev/queue';
import { emit, createConsumer } from '@gnldev/events';
import { scheduleWorkflow, pollScheduler } from '@gnldev/scheduler';

// ── queue: alerts are durable JOBS, not function calls ──────────────────────
// A crash mid-triage leaves the job pending, so it is picked up again instead of vanishing with the
// process that held it. The job id IS the incident id — a monitor that fires the same alert three
// times (they do) enqueues one job.
export const submitAlert = (alert: Alert) =>
  enqueue(storage.work, 'alert', alert, { id: `alert:${alert.incidentId}` });

const worker = createWorker(storage, {
  alert: async (payload) => { await incidentWorkflow.run(payload as Alert,
    { runId: `wf:${(payload as Alert).incidentId}`, journal: storage.runs }); },
});

// ── events: a restart publishes a FACT; whoever cares subscribes ────────────
// The agent does not know a watch exists, which is what lets you change the follow-up policy later
// without touching the prompt.
export const announceRestart = (incidentId: string, service: string) =>
  emit(storage.work, 'service.restarted', { incidentId, service }, { id: `restarted:${incidentId}` });

// ── scheduler: the 30-minute watch, as a row with a time on it ──────────────
// rb-memory promises something about a moment thirty minutes from now, made by a process that will
// very likely be redeployed before then. setTimeout keeps that promise only if nothing restarts.
const watcher = createConsumer(storage.work, 'service.restarted', async (p: any) => {
  await scheduleWorkflow(storage.runs, {
    id: `watch:${p.incidentId}`,           // idempotent on this id — a redelivered event schedules ONE watch
    name: 'post-restart-watch',
    input: { incidentId: p.incidentId, service: p.service },
    at: Date.now() + 30 * 60 * 1000,
  });
}, { name: 'post-restart-watcher' });

worker.start();
watcher.start();
// `.catch` is not decoration: a rejected promise from a bare interval is an unhandled rejection, and
// Node's default is to kill the process — so one transient poll error takes down alert intake too.
setInterval(() => { pollScheduler(storage.runs, runner).catch((e) => console.error('[sched]', e)); }, 5_000).unref();

// src/specialist.ts — a SECOND agent, with its own blast radius.
//
// rb-escalation says: "if a second opinion is needed, ask the database specialist rather than
// paging again." That is why this is a separate agent and not a longer system prompt — the
// specialist's tools reach a production database, and the triage agent runs on whatever pages it.
// One agent with both toolsets is one blast radius covering both.
import { createA2ATool } from '@gnldev/a2a';
import { createMcpTools, connectMcp } from '@gnldev/mcp';
import { createRestApi } from '@gnldev/server';

// MCP: the specialist's tools come from the org's database-inspection server, not from this repo.
// Calls made through it are journaled, so a resumed run does not re-query.
const client = await connectMcp({ type: 'stdio', command: 'db-mcp' });
const dbTools = await createMcpTools(client, { prefix: 'db_' });

const specialistApp = createRestApi({
  storage,
  agents: { dbSpecialist: { model: MODEL, tools: dbTools, maxSteps: 4,
    system: 'You are a database specialist. Answer with the query or pool number that explains the symptom.' } },
}, { title: 'DB Specialist', allowOpenAccess: true });

// A2A: the triage agent sees one more ordinary tool. Behind it is a POST to the specialist's REST
// API, wrapped so a timeout or a 500 becomes a failed tool result the triage model can read and
// react to — rather than an unhandled exception.
export const askSpecialist = createA2ATool({
  endpoint: process.env.SPECIALIST_URL ?? 'http://specialist.internal',
  agentName: 'dbSpecialist',
  description: 'Asks the database specialist for a second opinion. Use this instead of paging again.',
  timeoutMs: 10_000,
});

// src/server.ts — the surface a human touches.
import { Hono } from 'hono';
import { serve } from '@hono/node-server';
import { createRestApi } from '@gnldev/server';
import { createStudioApp } from '@gnldev/studio';
import { createChatRoute } from '@gnldev/chat-adapter';
import { roleAuth, makeGate } from '@gnldev/auth';
import { toJournal } from '@gnldev/durable';

// Two credential classes, because during an incident they are genuinely different people: everyone
// needs to read the journal immediately, and exactly one person may say yes to taking production
// down. One token for both means everyone who can watch can also restart.
const auth = roleAuth({
  admin: process.env.ONCALL_ADMIN ? { token: process.env.ONCALL_ADMIN } : undefined,
  viewer: process.env.ONCALL_VIEWER ? { token: process.env.ONCALL_VIEWER } : undefined,
});
const gate = makeGate(auth, { allowOpenAccess: !auth });

const app = new Hono();

app.post('/alerts', async (c) => {
  const body = await c.req.json();
  const jobId = await submitAlert(body);
  // 202, not 200: nothing has been triaged yet. Answering 200 with an empty diagnosis is how a
  // monitoring system learns to stop trusting the acknowledgement.
  return c.json({ accepted: true, jobId }, 202);
});

// The ONLY endpoint that can release a suspended restart. Note what is NOT here: any way to say
// "run the restart" — only a way to release a decision the agent already reached and the journal
// already recorded, addressed by that exact toolCallId.
app.post('/incidents/:runId/approve', async (c) => {
  if (!(await gate.allow(c.req.raw, 'write'))) return gate.deny(c.req.raw, 'write');
  const { toolCallId, approved } = await c.req.json();
  const r = await resume(c.req.param('runId'), { [toolCallId]: approved !== false });
  return c.json({ text: r.text, interrupts: r.interrupts });
});

// A streaming route the AI SDK's useChat talks to directly.
app.route('/chat', createChatRoute({ gnl }, { resolveThreadId: (_c, b) => b?.incidentId }));

// The generic REST API over agents/runs/workflows, and the journal inspector.
app.mount('/studio', createStudioApp({ reader: toJournal(storage.runs), apiBase: '/studio', auth }));
app.mount('/api', createRestApi({ storage, agents }, { auth, allowOpenAccess: !auth }));

// Loopback unless an address is named. A bare serve() binds every interface, which for an approval
// endpoint with no token is an admin surface offered to the network.
serve({ fetch: app.fetch, port: 3000, hostname: process.env.HOST ?? '127.0.0.1' });

// Express, Fastify, Koa and bare node:http go through one adapter.
import express from 'express';
import { toNodeHandler } from '@gnldev/server/node';
import { api, studio } from './gnl.js';

const server = express();

// YOUR routes first, with the body parser scoped to them.
//
// The one mistake worth naming: a GLOBAL express.json() mounted BEFORE GNL drains the request
// stream and hands the result to Express, not to us — the handler then sees a POST with no body
// and answers "runId is required" to a request that plainly carried one.
server.use('/app', express.json(), yourRouter);

server.use('/studio', toNodeHandler(studio));
server.use('/agents', toNodeHandler(api));

server.listen(3000, '127.0.0.1');

// src/observe.ts — did it follow the runbook, and what did it cost?
//
// Both answered from the JOURNAL after the fact, not from log lines the agent was asked to emit. A
// finished run can be scored, re-scored under a stricter rule later, and traced — because what
// happened is a data structure, not a transcript that existed once.
import { trajectoryScorerFor, scoreRun } from '@gnldev/evals';
import { exportRun } from '@gnldev/otel';
import { piiTextRedactor } from '@gnldev/processors';
import { toJournal } from '@gnldev/durable';

// The runbook, in a form a build can fail on. Deliberately about the SHAPE of the run, not the
// wording of the answer: a diagnosis can be phrased a hundred ways, but "called restartService
// during a latency incident" either happened or it did not.
export const RUNBOOK_RULES = {
  memory:  { requiredTools: ['searchRunbook', 'getMetric'], maxToolCalls: 8 },
  latency: { requiredTools: ['searchRunbook'], forbiddenTools: ['restartService'], maxToolCalls: 8 },
};

export const scoreIncident = (runId: string, rules) => {
  const reader = toJournal(storage.runs);
  return scoreRun(reader, runId, [trajectoryScorerFor(reader, { ...rules, name: 'runbook-adherence' })]);
};

// `redact` matters here specifically: the run's verdict is written by a path NO processor sees, so a
// provider that quotes the offending input back inside a refusal puts that text into the span raw —
// and from there into a collector that is usually somebody else's.
export const traceIncident = (runId: string) =>
  exportRun(toJournal(storage.runs), runId, {
    endpoint: process.env.OTEL_EXPORTER_OTLP_ENDPOINT,
    serviceName: 'oncall-agent',
    redact: piiTextRedactor(),
  });