Cron vs Daemon Agents: Choose the Right AI Runtime

Your agent works locally. Now you need to decide how it should run in production. Should a scheduler start it, let it finish, and shut it down? Or should one process stay alive and wait for work?

The answer depends on four criteria: execution duration, state, latency, and operating cost. Use cron for bounded work with relaxed latency. Use a daemon for continuous, stateful, low-latency work. Use a hybrid when your workload contains both.

Cron vs Daemon Agents: Two Runtime Models

A cron job starts a new process on a schedule. The process performs a bounded task and exits. A daemon is one long-running process that operates continuously in the background. These definitions match both the traditional Unix comparison and the Koha documentation.

For an AI workload, a cron-style runtime looks like this:

import json
from datetime import datetime, timezone
from pathlib import Path


def collect_inputs() -> list[dict]:
    # Replace with approved API calls or responsible data collection.
    return [{"service": "api", "status": "healthy"}]


def run_agent(inputs: list[dict]) -> str:
    # Replace with your model and tool calls.
    return "\n".join(f"{item['service']}: {item['status']}" for item in inputs)


def main() -> None:
    inputs = collect_inputs()
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "body": run_agent(inputs),
    }

    output = Path("report.json")
    output.write_text(json.dumps(report, indent=2))
    print(f"Wrote {output}")


if __name__ == "__main__":
    main()

The scheduler invokes the script. The script reads its inputs, calls its tools, writes durable output, and returns an exit code.

A daemon reverses that lifecycle. It starts during deployment or system boot, then waits for requests, messages, or events. The following executable example uses SQLite as a minimal durable queue. Its enqueue command receives events, while its worker command polls the queue and handles SIGINT or SIGTERM without abandoning the current event:

import argparse
import asyncio
import json
import logging
import os
import signal
import sqlite3
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
)

DB_PATH = Path(os.environ.get("AGENT_QUEUE_DB", "agent-queue.db"))
POLL_INTERVAL_SECONDS = 1.0
RETRY_DELAY_SECONDS = 5.0
LEASE_SECONDS = 300.0


def connect() -> sqlite3.Connection:
    connection = sqlite3.connect(DB_PATH)
    connection.row_factory = sqlite3.Row
    return connection


def initialize_queue() -> None:
    with connect() as connection:
        connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS events (
                id TEXT PRIMARY KEY,
                payload TEXT NOT NULL,
                status TEXT NOT NULL DEFAULT 'pending',
                attempts INTEGER NOT NULL DEFAULT 0,
                available_at REAL NOT NULL
            );

            CREATE TABLE IF NOT EXISTS results (
                event_id TEXT PRIMARY KEY,
                result TEXT NOT NULL,
                completed_at TEXT NOT NULL
            );
            """
        )


def enqueue(event: dict[str, Any]) -> bool:
    event_id = event.get("id")
    if not isinstance(event_id, str) or not event_id:
        raise ValueError("The event must contain a non-empty string id")

    with connect() as connection:
        cursor = connection.execute(
            """
            INSERT OR IGNORE INTO events (id, payload, available_at)
            VALUES (?, ?, ?)
            """,
            (event_id, json.dumps(event), time.time()),
        )
        return cursor.rowcount == 1


def claim_next_event() -> dict[str, Any] | None:
    now = time.time()
    connection = connect()

    try:
        connection.execute("BEGIN IMMEDIATE")
        row = connection.execute(
            """
            SELECT id, payload
            FROM events
            WHERE
                (status = 'pending' AND available_at <= ?)
                OR
                (status = 'processing' AND available_at <= ?)
            ORDER BY available_at, id
            LIMIT 1
            """,
            (now, now),
        ).fetchone()

        if row is None:
            connection.commit()
            return None

        connection.execute(
            """
            UPDATE events
            SET status = 'processing',
                attempts = attempts + 1,
                available_at = ?
            WHERE id = ?
            """,
            (now + LEASE_SECONDS, row["id"]),
        )
        connection.commit()
        return json.loads(row["payload"])
    except Exception:
        connection.rollback()
        raise
    finally:
        connection.close()


async def handle(event: dict[str, Any]) -> dict[str, Any]:
    logging.info("Handling event %s", event["id"])

    # Replace this body with model and tool calls.
    await asyncio.sleep(0)
    return {
        "event_id": event["id"],
        "handled_at": datetime.now(timezone.utc).isoformat(),
    }


def mark_completed(event_id: str, result: dict[str, Any]) -> None:
    completed_at = datetime.now(timezone.utc).isoformat()

    with connect() as connection:
        connection.execute(
            """
            INSERT OR REPLACE INTO results (event_id, result, completed_at)
            VALUES (?, ?, ?)
            """,
            (event_id, json.dumps(result), completed_at),
        )
        connection.execute(
            "UPDATE events SET status = 'done' WHERE id = ?",
            (event_id,),
        )


def schedule_retry(event_id: str) -> None:
    with connect() as connection:
        connection.execute(
            """
            UPDATE events
            SET status = 'pending', available_at = ?
            WHERE id = ?
            """,
            (time.time() + RETRY_DELAY_SECONDS, event_id),
        )


async def worker() -> None:
    stop_requested = asyncio.Event()
    loop = asyncio.get_running_loop()

    def request_stop() -> None:
        if not stop_requested.is_set():
            logging.info("Shutdown requested; finishing the current event")
            stop_requested.set()

    for received_signal in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(received_signal, request_stop)
        except NotImplementedError:
            signal.signal(
                received_signal,
                lambda *_: loop.call_soon_threadsafe(request_stop),
            )

    while not stop_requested.is_set():
        event = claim_next_event()

        if event is None:
            try:
                await asyncio.wait_for(
                    stop_requested.wait(),
                    timeout=POLL_INTERVAL_SECONDS,
                )
            except TimeoutError:
                pass
            continue

        try:
            result = await handle(event)
            mark_completed(event["id"], result)
            logging.info("Completed event %s", event["id"])
        except Exception:
            logging.exception("Event %s failed; scheduling a retry", event["id"])
            schedule_retry(event["id"])

    logging.info("Worker stopped cleanly")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command", required=True)

    enqueue_parser = subparsers.add_parser("enqueue")
    enqueue_parser.add_argument(
        "event",
        help='JSON event such as \'{"id":"evt-1","message":"hello"}\'',
    )

    subparsers.add_parser("worker")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    initialize_queue()

    if args.command == "enqueue":
        event = json.loads(args.event)
        accepted = enqueue(event)
        print("Event enqueued" if accepted else "Duplicate event ignored")
    else:
        asyncio.run(worker())


if __name__ == "__main__":
    main()

Save the example as daemon.py, start the consumer, and enqueue an event from another terminal:

python daemon.py worker
python daemon.py enqueue '{"id":"evt-1","message":"hello"}'

SQLite preserves queued events across process restarts. The lease makes an event eligible for processing again if a worker crashes after claiming it. In production, the same lifecycle can use a managed durable queue, and a process supervisor should restart the worker after unexpected exits.

This process does not exit after one task. It needs supervision and health checks, while its signal handler provides a graceful shutdown strategy.

The distinction is about lifecycle, not intelligence. The same LangChain, LangGraph, CrewAI, or custom Python workflow can run under either model.

The Decision Matrix: Execution Duration and State

Start with how long the work lasts and what must survive between runs.

Criterion	Choose cron when…	Choose a daemon when…
Execution duration	Work is autonomous, bounded, and known before execution	Work is continuous or invoked very frequently
State	Each run can reconstruct context from durable storage	The process benefits from cached context, persistent sessions, or in-memory state

Execution duration

Cron fits work with a clear beginning and end:

• Generate a periodic product report.
• Enrich a queued dataset.
• Check approved sources for content changes.
• Summarize application logs.
• Send a digest, then exit.

A daemon fits work that does not have a natural final step. It may listen on a socket, consume events, or maintain an interactive session. The Unix guidance also favors a daemon when work occurs more than a few times per hour or even more frequently than once per minute, though the exact crossover depends on the workload.

Do not confuse a long task with a continuous task. A batch agent can run for an extended period and still be conceptually bounded. If it processes a known input set, saves the result, and exits, a scheduled job can remain the simpler model.

For a concrete way to structure bounded tool calls, retries, and output steps, see this guide to scheduled multi-step AI workflows.

State

A daemon can retain state in memory. That makes it useful for:

• Cached reference data.
• Persistent client sessions.
• Deferred disk writes.
• Reused network connections.
• Conversation context needed across interactions.

Its longevity is both an advantage and a liability. In-memory state disappears when the process crashes or is replaced. Treat it as a cache unless you can safely reconstruct it.

A cron job should assume a clean process on every run. Store checkpoints in a database or object store. This complete example uses a JSON file as a minimal durable store and resumes a list of items after the last saved cursor:

import json
from pathlib import Path
from typing import Any

STATE_PATH = Path("batch-state.json")


class JsonFileStore:
    def __init__(self, path: Path) -> None:
        self.path = path

    def _load(self) -> dict[str, Any]:
        if not self.path.exists():
            return {"checkpoint": {"cursor": None}, "results": {}}
        return json.loads(self.path.read_text())

    def _save(self, state: dict[str, Any]) -> None:
        temporary_path = self.path.with_suffix(".tmp")
        temporary_path.write_text(json.dumps(state, indent=2))
        temporary_path.replace(self.path)

    def load_checkpoint(self) -> dict[str, str | None]:
        return self._load()["checkpoint"]

    def save_result(self, item_id: str, result: dict[str, Any]) -> None:
        state = self._load()
        state["results"][item_id] = result
        self._save(state)

    def save_checkpoint(self, checkpoint: dict[str, str | None]) -> None:
        state = self._load()
        state["checkpoint"] = checkpoint
        self._save(state)


def process(item: dict[str, Any]) -> dict[str, Any]:
    return {
        "source": item["name"],
        "normalized_value": str(item["value"]).strip().lower(),
    }


def run_batch(
    store: JsonFileStore,
    items: list[dict[str, Any]],
) -> None:
    checkpoint = store.load_checkpoint()
    cursor = checkpoint["cursor"]
    start_index = 0

    if cursor is not None:
        item_ids = [item["id"] for item in items]
        try:
            start_index = item_ids.index(cursor) + 1
        except ValueError as error:
            raise ValueError(
                f"Checkpoint cursor {cursor!r} is not present in the input"
            ) from error

    for item in items[start_index:]:
        result = process(item)
        store.save_result(item["id"], result)
        store.save_checkpoint({"cursor": item["id"]})
        print(f"Processed {item['id']}")


def main() -> None:
    items = [
        {"id": "item-001", "name": "First", "value": " READY "},
        {"id": "item-002", "name": "Second", "value": " PENDING "},
        {"id": "item-003", "name": "Third", "value": " DONE "},
    ]
    run_batch(JsonFileStore(STATE_PATH), items)


if __name__ == "__main__":
    main()

This design makes restarts explicit. It also prevents your agent’s memory from depending on one process surviving forever.

The Decision Matrix: Latency and Operating Cost

The next question is how quickly work must begin and what you are willing to keep running.

Criterion	Choose cron when…	Choose a daemon when…
Latency	Minutes-level freshness is acceptable	Requests or events need near-immediate handling
Operating cost	You want resources allocated per execution	You accept idle compute and supervision for persistent availability

Latency

A scheduled agent can only react when its next invocation starts. That is usually fine for reports, digests, backups, periodic enrichment, and routine monitoring.

Use a daemon when another system expects an available service. Common examples include:

• Interactive chat.
• Streaming model output.
• Webhook handling.
• Continuous queue consumption.
• High-frequency tool loops.
• Persistent sessions with external clients.

A webhook does not always require a monolithic daemon. You can separate the always-available receiver from the agent worker. The receiver validates the request and writes an event to durable storage. A worker then processes it independently. This keeps low ingress latency without forcing every part of the workflow into one long-lived process.

Operating cost

A daemon consumes memory while idle. Long-running processes are also more exposed to memory leaks. If a daemon exits after an error, it will not restart unless you implement that behavior or use a supervisor.

Cron isolates executions. A failed run does not inherently keep a damaged process alive for the next run. This is why the cited Unix comparison generally favors cron for robustness and daemons for performance, with overlap between the models.

Cron still has overhead. Every invocation may need to:

• Start the runtime.
• Load dependencies.
• Read configuration and secrets.
• Restore state.
• Establish API connections.
• Initialize model clients.

A daemon can amortize that setup. But you pay for persistent availability and must operate the service around it.

Compare the full workload rather than only model-token spend. The production AI agent cost breakdown covers the wider self-hosted, managed, and serverless tradeoffs. For bounded jobs, serverless AI agent deployment also explains the pay-per-execution model.

Apply the Matrix to Common AI Agent Workloads

Batch reports are the clearest cron workload. Inputs are known when execution begins. Results can go to a database, file, email system, or internal API. Minutes-level latency is normally acceptable.

The same applies to responsible scraping, data enrichment, notifications, and backups. The Koha documentation gives practical scheduled examples: daily backups, sitemap generation, hold-queue construction every one to four hours, and daily cancellation of expired holds.

Indexing shows where frequency changes the answer. Koha documents an index-rebuild cron job scheduled every five to fifteen minutes according to performance needs. Newer installations described in the documentation replaced it with the koha-indexer daemon, which indexes new and modified data every thirty seconds. The workload moved from periodic freshness toward continuous processing.

For AI agents, apply the same reasoning:

Workload	Likely runtime	Main reason
Periodic competitive summary	Cron	Bounded work and relaxed freshness
Dataset enrichment	Cron	Durable input and output state
Daily operational report	Cron	Predictable schedule
Interactive support assistant	Daemon	Session state and low latency
Continuous event monitor	Daemon	Ongoing consumption
Webhook-triggered triage	Daemon or hybrid	Immediate intake with durable processing

Whichever runtime you select, add four production controls.

Idempotency: The same input may arrive again. Use a stable event or job identifier. Before creating an external side effect, check whether that identifier has already completed.

Durable checkpoints: Persist progress after successful units of work. Do not rely exclusively on an in-memory list of completed steps.

Concurrency control: Prevent overlapping cron invocations from processing the same records. For daemons, define how many workers may claim work and make the claim atomic.

Failure visibility: Emit structured logs with the job ID, event ID, tool name, and final status. Cron failures need alerts. Daemons need both alerts and supervised restarts.

Three AI Runtime Architectures to Choose From

1. Scheduled serverless AI job

Use this for bounded, infrequent, cost-sensitive work.

A scheduler invokes your Python entry point. The job loads durable state, runs the workflow, stores its outputs, and exits. Secrets come from the runtime rather than source control.

• Duration: Bounded.
• State: Reloaded from durable storage.
• Latency: Schedule-based.
• Operating cost: Resources are needed during execution rather than while waiting.

This is a strong default for reports, enrichment, monitoring summaries, and periodic API integrations. The serverless agent deployment model is a natural fit when you do not need a permanent listener.

2. Always-on daemon on a managed VM

Use this for interactive, stateful, low-latency work.

Run the agent as a supervised service. Give it health checks, graceful shutdown handling, structured logs, bounded concurrency, and durable recovery state.

• Duration: Continuous.
• State: Cached in memory and backed by durable storage.
• Latency: Near-immediate.
• Operating cost: Includes idle resources and service supervision.

This fits chat sessions, streaming responses, continuous consumers, and agents that must remain available to other programs.

3. Hybrid scheduler and event service

Use this when one runtime cannot satisfy the entire workflow.

Keep a daemon available for events or sessions. Move expensive, retryable, or periodic work into durable jobs. For example, a webhook service can validate and enqueue an event immediately, while a scheduled job builds the broader report later.

• Duration: Continuous ingress plus bounded workers.
• State: Durable queue and checkpoints, with optional in-memory session caches.
• Latency: Immediate intake with asynchronous completion.
• Operating cost: Persistent compute only where availability requires it.

This pattern is more complex, but it keeps lifecycle decisions local. Your event receiver does not need to become your batch processor. Your scheduled workflow does not need to pretend it is an interactive service.

Cron versus daemon is not a framework choice. It is a workload choice. Start with duration, state, latency, and operating cost. Prefer a scheduled job when work can wake, run, persist, and exit. Keep a daemon when the agent must remain available or retain active sessions. Split the system when both are true.

HollowHost’s scheduled and serverless deployment model fits the first side of that decision. Even if your final architecture is hybrid, moving bounded agent work out of an always-on process can make the system easier to operate and recover.