> Level 16 of Agent Harness Patterns, a track of patterns on how AI agents work. Web version: https://harnesspatterns.dev/patterns/proactive-agents · All patterns: https://harnesspatterns.dev/llms.txt

# Proactive agents

Every agent so far waited for someone to type. A proactive one wakes up on a timer, looks around, and speaks up only when there’s something worth saying.

## The problem

Some jobs have no moment when a person would think to ask: a pet that gets hungry while its owner is at school, an order that gets stuck, a server that starts failing at night. An agent that only answers when spoken to is no use for them.

The obvious fix is a timer that runs the agent every so often. Done carelessly, that’s the Unhinged Cuckoo: every tick wakes the model, every run costs tokens, and every run messages you “all fine”. By the third message you stop reading them, and the one that mattered goes unread.

## The solution

A heartbeat: a timer that hands the agent a fixed prompt, the `checkPrompt`, as if someone had typed it. What makes it useful instead of noisy is what you put around that timer:

- **Check before you wake**
   localCondition
   Plain code that runs on every tick, before the model: read a gauge, a file, a row. While it says no, the tick costs nothing.
- **One run at a time**
   built in
   A tick that fires while the last run is still going is dropped, not queued. Two runs never share the history at once.
- **Fuses**
   maxTicks, timeoutMs, runTimeoutMs
   A budget of ticks, of wall-clock time, and of time for each run, so it ends even if you forget to stop it.
- **Speak once, and switch off**
   stopHeartbeat()
   Message the person only when something happened, with how it ended. When the job is over, turn the heartbeat off.

The first one does most of the work. Most ticks find nothing to do, and deciding that doesn’t take a model: it takes an `if`. In the animation, five ticks go by and only one of them calls the model.

## The cast

Same cast as always, inside a pocket pet.

- **The clock** (the heartbeat): Its bell rings every two hours, and stays lit while the heartbeat is on.
- **The bracket** (localCondition): Blinks around the hearts on every tick: your own code reading the gauges. No model involved.
- **Astor** (the loop): Naps on his mat until a tick says a gauge is low, then runs the loop as usual.
- **The Oracle** (the model): Asleep until Astor brings it the checkPrompt.
- **The icons** (the tools): Status, food, game and the call light: `check_status`, `feed`, `play` and `beep_owner`.
- **The owner** (the person): At school all day. Gets one beep, and it’s already good news.

In the EventBus panel, the quiet ticks are only your code: astorlm emits nothing for a tick your check turned down. `heartbeat_tick` shows up once, when the model is actually woken.

## The code

**With astorlm:** pass `heartbeat` to the agent and it starts on its own. The guards are options: `localCondition`, `maxTicks` and `timeoutMs`; overlapping ticks are dropped for you. Your app calls `stopHeartbeat()` when the owner is back.

**From scratch:** a timer around the loop from level 2. A flag keeps runs from overlapping, a counter and a deadline are the fuses, and a plain function decides whether the model is called at all.

**With astorlm**

```ts
import { OpenAIProvider, createLocalAgent, tool } from 'astorlm'
import { z } from 'zod'

const HOUR = 60 * 60_000

const checkStatus = tool({
  name: 'check_status',
  description: 'Open the status screen: hunger and happiness in hearts, and whether the pet is sick or asleep.',
  schema: z.object({}),
  execute: async () => pet.status(), // your code: the pet lives in your app
})

const feed = tool({
  name: 'feed',
  description: 'Feed the pet a meal or a snack. A meal fills hunger; a snack only cheers it up.',
  schema: z.object({ food: z.enum(['meal', 'snack']) }),
  execute: async ({ food }) => pet.feed(food),
})

const play = tool({
  name: 'play',
  description: 'Play the left-or-right game with the pet. Winning fills happiness.',
  schema: z.object({}),
  execute: async () => pet.play(),
})

const beepOwner = tool({
  name: 'beep_owner',
  description: 'Beep the owner with a short message. They are at school: only when something happened.',
  schema: z.object({ text: z.string() }),
  execute: async ({ text }) => {
    await sendPush(text) // your code
    return 'Beeped.'
  },
})

const agent = await createLocalAgent({
  // Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
  provider: new OpenAIProvider({
    baseURL: 'http://localhost:11434/v1', // e.g. Ollama's default address
    model: 'your-model', // e.g. 'llama3.1', 'gpt-4o-mini'
    apiKey: 'YOUR_API_KEY', // local servers usually ignore it
  }),
  tools: [checkStatus, feed, play, beepOwner],
  maxTurns: 8,
  // Starts on its own as soon as the agent is created. Nobody types anything.
  heartbeat: {
    intervalMs: 2 * HOUR,
    checkPrompt: 'Check on Milonga. Take care of whatever is low, then beep her owner with one line.',
    // Runs on every tick, before the model. While it says no, a tick costs 0 tokens.
    localCondition: () => pet.hunger <= 1 || pet.happy <= 1,
    maxTicks: 6, // the fuses: a school day of ticks at most…
    timeoutMs: 10 * HOUR, // …and of wall-clock time
  },
})

// The owner is home: the app takes over and switches the heartbeat off.
onOwnerHome(() => agent.stopHeartbeat())

// Quiet ticks emit nothing. The ones that wake the model do:
agent.on('event', (event) => {
  if (event.type === 'heartbeat_tick') console.log('heartbeat woke the agent')
})
```

**TypeScript**

```ts
// A heartbeat, from scratch. Plain fetch and timers, no SDK.

// Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
const LLM = {
  baseURL: 'http://localhost:11434/v1', // e.g. Ollama's default address
  model: 'your-model', // e.g. 'llama3.1', 'gpt-4o-mini'
  apiKey: 'YOUR_API_KEY', // local servers usually ignore it
}

const HOUR = 60 * 60_000

// 1. The tools, as in level 3. The pet lives in your app.
type ToolFn = (args: Record<string, string | number>) => Promise<string>
const tools: Record<string, ToolFn> = {
  check_status: async () => pet.status(),
  feed: async ({ food }) => pet.feed(String(food)),
  play: async () => pet.play(),
  beep_owner: async ({ text }) => {
    await sendPush(String(text)) // your code
    return 'Beeped.'
  },
}
const toolSchemas = [/* one JSON Schema per tool: check_status(), feed(food), play(), beep_owner(text) */]

type ToolCall = { id: string; function: { name: string; arguments: string } }
type Message =
  | { role: 'user'; content: string }
  | { role: 'assistant'; content: string | null; tool_calls?: ToolCall[] }
  | { role: 'tool'; tool_call_id: string; content: string }

// 2. The loop from level 2, unchanged. Each tick that gets through is one run of it.
async function runAgent(prompt: string, maxTurns = 8): Promise<string> {
  const messages: Message[] = [{ role: 'user', content: prompt }]
  for (let turn = 1; turn <= maxTurns; turn++) {
    const res = await fetch(`${LLM.baseURL}/chat/completions`, {
      method: 'POST',
      headers: { 'content-type': 'application/json', authorization: `Bearer ${LLM.apiKey}` },
      body: JSON.stringify({ model: LLM.model, messages, tools: toolSchemas }),
    })
    const [choice] = (await res.json()).choices
    const reply: Message = choice.message
    messages.push(reply)
    if (choice.finish_reason !== 'tool_calls') return reply.content ?? ''

    for (const call of reply.tool_calls ?? []) {
      const run = tools[call.function.name]
      let output = `Unknown tool: ${call.function.name}`
      try {
        if (run) output = await run(JSON.parse(call.function.arguments))
      } catch (err) {
        output = `Error: ${err instanceof Error ? err.message : err}`
      }
      messages.push({ role: 'tool', tool_call_id: call.id, content: output })
    }
  }
  throw new Error(`No answer after ${maxTurns} turns`)
}

// 3. The heartbeat: a timer, a cheap check before the model, and fuses.
const CHECK_PROMPT = 'Check on Milonga. Take care of whatever is low, then beep her owner with one line.'
const MAX_TICKS = 6

// Plain code, no model: while it says no, a tick costs 0 tokens.
const localCondition = (): boolean => pet.hunger <= 1 || pet.happy <= 1

let running = false
let ticks = 0

async function tick(): Promise<void> {
  if (running) return // still busy with the last tick: skip this one, never overlap
  if (++ticks > MAX_TICKS) return stop() // fuse: a budget of ticks
  if (!localCondition()) return // nothing low: let the model sleep

  running = true
  try {
    console.log(await runAgent(CHECK_PROMPT))
  } catch (err) {
    console.error('heartbeat run failed:', err) // log it, and let the next tick try again
  } finally {
    running = false
  }
}

const timer = setInterval(tick, 2 * HOUR)
const deadline = setTimeout(stop, 10 * HOUR) // fuse: wall-clock time

function stop(): void {
  clearInterval(timer)
  clearTimeout(deadline)
}

// The owner is home: the app takes over.
onOwnerHome(stop)
```

**Python**

```python
# A heartbeat, from scratch. Standard library only, no SDK.
import json
import threading
import time
import urllib.request

# Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy...
LLM = {
    "base_url": "http://localhost:11434/v1",  # e.g. Ollama's default address
    "model": "your-model",  # e.g. "llama3.1", "gpt-4o-mini"
    "api_key": "YOUR_API_KEY",  # local servers usually ignore it
}

HOUR = 60 * 60

def post(path, payload):
    request = urllib.request.Request(
        f"{LLM['base_url']}{path}",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {LLM['api_key']}"},
    )
    with urllib.request.urlopen(request) as response:
        return json.load(response)

# 1. The tools, as in level 3. The pet lives in your app.
def beep_owner(text):
    send_push(text)  # your code
    return "Beeped."

TOOLS = {
    "check_status": lambda: pet.status(),
    "feed": lambda food: pet.feed(food),
    "play": lambda: pet.play(),
    "beep_owner": beep_owner,
}
TOOL_SCHEMAS = [...]  # one JSON Schema per tool: check_status(), feed(food), play(), beep_owner(text)

# 2. The loop from level 2, unchanged. Each tick that gets through is one run of it.
def run_agent(prompt, max_turns=8):
    messages = [{"role": "user", "content": prompt}]

    for _ in range(max_turns):
        choice = post("/chat/completions", {"model": LLM["model"], "messages": messages, "tools": TOOL_SCHEMAS})["choices"][0]
        reply = choice["message"]
        messages.append(reply)
        if choice["finish_reason"] != "tool_calls":
            return reply.get("content") or ""

        for call in reply.get("tool_calls", []):
            try:
                output = TOOLS[call["function"]["name"]](**json.loads(call["function"]["arguments"]))
            except Exception as err:
                output = f"Error: {err}"
            messages.append({"role": "tool", "tool_call_id": call["id"], "content": output})

    raise RuntimeError(f"No answer after {max_turns} turns")

# 3. The heartbeat: a timer, a cheap check before the model, and fuses.
CHECK_PROMPT = "Check on Milonga. Take care of whatever is low, then beep her owner with one line."
INTERVAL = 2 * HOUR
MAX_TICKS = 6
DEADLINE = time.monotonic() + 10 * HOUR  # fuse: wall-clock time
stopped = threading.Event()
on_owner_home(stopped.set)  # the owner is home: the app takes over

def local_condition():
    """Plain code, no model: while it says no, a tick costs 0 tokens."""
    return pet.hunger <= 1 or pet.happy <= 1

# One thread, one run at a time: a tick can never overlap the last one.
ticks = 0
while not stopped.wait(INTERVAL):
    ticks += 1
    if ticks > MAX_TICKS or time.monotonic() > DEADLINE:
        break  # the fuses
    if not local_condition():
        continue  # nothing low: let the model sleep
    try:
        print(run_agent(CHECK_PROMPT))
    except Exception as err:
        print("heartbeat run failed:", err)  # log it, and let the next tick try again
```

## What to watch

- **astorlm’s heartbeat keeps one session.** Every tick that wakes the model adds to the same history, so a heartbeat that wakes it often grows its context, and its bill, with every run. Keep the checkPrompt short, add compaction, or start each run on a fresh agent (the from-scratch versions above do).
- **Mind the per-run timeout.** `runTimeoutMs` defaults to 60 seconds. A run whose tools wait on something slow needs more, or it will be aborted halfway.
- **A heartbeat is not a cron job.** It lives in your process: if the process stops, so does the heartbeat, and the ticks it missed are gone. For jobs that must survive restarts, let a real scheduler start the agent, and keep the same guards.
- **Decide what’s worth a message before you write the prompt.** “Beep me if you had to do something”, not “tell me how it’s going”. A message that says nothing trains the person to ignore the next one.
- **Acting alone still needs limits.** Nobody is watching. Feeding the pet is fine; anything you can’t take back should wait for a person’s yes.

## Related patterns

- [4 · When to stop](https://harnesspatterns.dev/patterns/when-to-stop.md)
- [7 · The backpack fills up](https://harnesspatterns.dev/patterns/compaction.md)
- [10 · Fresh laps](https://harnesspatterns.dev/patterns/fresh-laps.md)
- [13 · Human in the loop](https://harnesspatterns.dev/patterns/human-in-the-loop.md)
- [11 · Observability and evals](https://harnesspatterns.dev/patterns/observability.md)