Build an Always-On Personal Agent Workflow with Cloud Continuity & Spend Caps

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

On July 25, 2026, Google moved Gemini Spark from the $99.99 Ultra tier down to the $19.99 AI Pro plan for US users. That single pricing change is worth more attention than most model releases: it turned the always-on personal agent from an expensive novelty into a consumer product. Spark is a 24/7 agent that runs on Google's cloud, works even when your phone and laptop are off, and acts under your direction. This dispatch builds spark-sleeper, a LangGraph workflow that gives you the same pattern as an open system you control — a persistent personal agent with a durable task queue, schedule and trigger wake-ups, cloud continuity, and a hard monthly spend cap enforced before every model call. The latest AI news hub tracked the always-on agent wave; this is the workflow underneath it.

Why always-on changes the agent economics

The shift from session-based to always-on agents is not cosmetic. A session agent is a function call: you invoke it, it answers, it dies. An always-on agent is a worker with a queue: it holds a backlog of tasks, wakes when it should, executes for hours if needed, and reports when done. That changes three things. First, continuity — the agent survives your laptop closing, because it runs on cloud compute, not on your device. Second, duration — long-horizon tasks like nightly report generation, price monitoring, or multi-step research become practical because the agent can run for hours with checkpointing. Third, economics — an agent that runs 24/7 is a recurring cost, so spend governance stops being optional and becomes the core design constraint. Gemini Spark's pricing made the first two mainstream; the third is what this workflow is built around.

Architecture

flowchart TD
    A[User device: enqueue task] --> B[Task queue store]
    C[Scheduler: cron / trigger] --> B
    B --> D[Wake + load task]
    D --> E[Budget check: spend vs monthly cap]
    E -- over cap --> F[Pause + notify user]
    E -- ok --> G[Executor: run task with checkpointing]
    G --> H{More steps?}
    H -- yes --> E
    H -- no --> I[Reporter: push result to channel]
    I --> J[Update task store + trail]
    J --> K[Sleep until next wake]

Project setup

mkdir spark-sleeper && cd spark-sleeper
python -m venv .venv && source .venv/bin/activate
pip install langgraph langchain-openai pydantic apscheduler

# .env
OPENAI_API_KEY=sk-...
MODEL=openai/gpt-5.6-luna         # cheap workhorse for always-on loops
MODEL_HARD=openai/gpt-5.6-sol     # expensive reasoning, used sparingly
MONTHLY_CAP_USD=20.0
MAX_RUN_COST_USD=2.0
TASK_STORE_PATH=./data/tasks.jsonl
CHECKPOINT_DIR=./checkpoints
NOTIFY_CHANNEL=telegram            # or slack / email
WAKE_CRON=0 2 * * *                # nightly maintenance run

schemas.py

from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime

class Task(BaseModel):
    id: str
    kind: Literal["research", "report", "monitor", "reminder"]
    prompt: str
    source_device: str = "web"
    status: Literal["queued", "running", "done", "paused", "failed"] = "queued"
    created_at: datetime = Field(default_factory=datetime.utcnow)
    checkpoint: dict = Field(default_factory=dict)

class SpendRecord(BaseModel):
    task_id: str
    model: str
    tokens_in: int = 0
    tokens_out: int = 0
    cost_usd: float = 0.0
    at: datetime = Field(default_factory=datetime.utcnow)

class BudgetState(BaseModel):
    month_key: str
    spent_usd: float = 0.0
    cap_usd: float = 20.0
    over: bool = False

class RunResult(BaseModel):
    task_id: str
    summary: str
    cost_usd: float
    finished_at: datetime = Field(default_factory=datetime.utcnow)

tools.py

import os
import json
import datetime
from schemas import Task, SpendRecord, BudgetState

TASK_PATH = os.getenv("TASK_STORE_PATH", "./data/tasks.jsonl")
CHK_DIR = os.getenv("CHECKPOINT_DIR", "./checkpoints")

async def enqueue_task(task: Task) -> Task:
    os.makedirs(os.path.dirname(TASK_PATH), exist_ok=True)
    with open(TASK_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(task.model_dump()) + "
")
    return task

async def load_ready_task() -> Task | None:
    # Read queue, return oldest queued task
    try:
        with open(TASK_PATH, encoding="utf-8") as f:
            for line in f:
                t = Task(**json.loads(line))
                if t.status == "queued":
                    return t
    except FileNotFoundError:
        return None
    return None

async def update_task(task: Task):
    # Rewrite queue with updated status (simplified; use a real DB in prod)
    lines = []
    try:
        with open(TASK_PATH, encoding="utf-8") as f:
            lines = f.readlines()
    except FileNotFoundError:
        return
    with open(TASK_PATH, "w", encoding="utf-8") as f:
        for line in lines:
            t = Task(**json.loads(line))
            if t.id == task.id:
                f.write(json.dumps(task.model_dump()) + "
")
            else:
                f.write(line)

async def read_budget(month_key: str) -> BudgetState:
    # Track spend in a small JSON ledger; a real deployment uses a DB
    ledger = os.path.join(os.path.dirname(TASK_PATH), "budget.json")
    try:
        with open(ledger, encoding="utf-8") as f:
            data = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        data = {}
    b = data.get(month_key, {"spent_usd": 0.0})
    return BudgetState(month_key=month_key, spent_usd=b["spent_usd"], cap_usd=float(os.getenv("MONTHLY_CAP_USD", "20")))

async def record_spend(rec: SpendRecord):
    ledger = os.path.join(os.path.dirname(TASK_PATH), "budget.json")
    try:
        with open(ledger, encoding="utf-8") as f:
            data = json.load(f)
    except (FileNotFoundError, json.JSONDecodeError):
        data = {}
    month_key = datetime.datetime.utcnow().strftime("%Y-%m")
    entry = data.get(month_key, {"spent_usd": 0.0})
    entry["spent_usd"] += rec.cost_usd
    data[month_key] = entry
    with open(ledger, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2)

def save_checkpoint(task_id: str, state: dict):
    os.makedirs(CHK_DIR, exist_ok=True)
    with open(os.path.join(CHK_DIR, f"{task_id}.json"), "w", encoding="utf-8") as f:
        json.dump(state, f)

def load_checkpoint(task_id: str) -> dict:
    try:
        with open(os.path.join(CHK_DIR, f"{task_id}.json"), encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

graph.py

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from schemas import Task, BudgetState, RunResult
from tools import load_ready_task, update_task, read_budget, record_spend, save_checkpoint, load_checkpoint
from datetime import datetime

class SleeperState(TypedDict):
    task: Task
    budget: BudgetState
    result: RunResult
    step: int

async def wake_node(state: SleeperState) -> SleeperState:
    task = await load_ready_task()
    if task is None:
        return {**state, "task": None}
    task.status = "running"
    await update_task(task)
    return {**state, "task": task, "step": 0}

async def budget_node(state: SleeperState) -> SleeperState:
    b = await read_budget(datetime.utcnow().strftime("%Y-%m"))
    b.over = b.spent_usd >= b.cap_usd
    return {**state, "budget": b}

def route_budget(state: SleeperState) -> Literal["execute", "pause"]:
    if not state.get("task"):
        return "pause"  # nothing to do, sleep
    return "pause" if state["budget"].over else "execute"

async def execute_node(state: SleeperState) -> SleeperState:
    task = state["task"]
    # Run one step of the task (a real impl calls the model); checkpoint between steps
    cp = load_checkpoint(task.id)
    step = cp.get("step", 0)
    # ... call model here, accumulate step result ...
    save_checkpoint(task.id, {**cp, "step": step + 1})
    await record_spend(SpendRecord(task_id=task.id, model="gpt-5.6-luna", tokens_in=500, tokens_out=200, cost_usd=0.012))
    done = step >= 4  # example: task completes after 5 steps
    if done:
        task.status = "done"
        await update_task(task)
        return {**state, "task": task, "result": RunResult(task_id=task.id, summary="Task completed", cost_usd=0.06)}
    return {**state, "task": task, "step": step + 1}

async def pause_node(state: SleeperState) -> SleeperState:
    # Over cap: mark task paused, notify user, do not spend
    if state.get("task"):
        state["task"].status = "paused"
        await update_task(state["task"])
    return {**state}

async def report_node(state: SleeperState) -> SleeperState:
    # Push result summary to NOTIFY_CHANNEL (telegram/slack/email)
    return {**state}

def build_graph():
    g = StateGraph(SleeperState)
    g.add_node("wake", wake_node)
    g.add_node("budget", budget_node)
    g.add_node("execute", execute_node)
    g.add_node("pause", pause_node)
    g.add_node("report", report_node)
    g.set_entry_point("wake")
    g.add_edge("wake", "budget")
    g.add_conditional_edges("budget", route_budget, {"execute": "execute", "pause": "pause"})
    g.add_edge("execute", "report")
    g.add_edge("report", END)
    g.add_edge("pause", END)
    return g.compile()

main.py

import asyncio
from graph import build_graph, SleeperState
from schemas import Task

async def main():
    graph = build_graph()
    # Simulate one wake cycle: a queued task exists, budget is under cap
    state = await graph.ainvoke({"task": Task(id="t1", kind="report", prompt="Nightly summary"), "step": 0})
    print(f"task status: {state.get('task').status if state.get('task') else 'none'}")
    print(f"budget spent: {state['budget'].spent_usd:.2f} / {state['budget'].cap_usd}")

if __name__ == "__main__":
    asyncio.run(main())

Retry rules

• Task-queue reads retry twice with backoff on transient file/DB errors; a missing task is a normal empty queue, not an error.
• Model calls retry twice with exponential backoff (1s, 2s) on 5xx; each retry is metered against MAX_RUN_COST_USD and stops the task if the per-run cap is hit.
• Checkpoint writes are critical-path: if a checkpoint cannot be saved, the step is not considered complete — resumability is a guarantee, not a best effort.
• Budget reads are cached for 60 seconds; budget writes are serialized so concurrent wakes cannot double-spend past the cap.
• A paused task is never retried automatically while over cap; it stays paused until the user tops up or the next month resets the ledger.

The scheduler and the queue

The always-on pattern rests on two pieces that look boring and matter enormously: the scheduler and the queue. The scheduler decides when the agent wakes — a cron expression for nightly maintenance, a webhook for an external trigger, or a manual "do this now" from a device. The queue decouples the user's request from the agent's execution: enqueueing is instant and device-friendly, while execution happens on the agent's clock. That decoupling is what makes cloud continuity possible — the user's phone can drop offline a second after enqueueing and the task still completes, because the task lives in the queue, not on the device. The same pattern shows up in every durable-agent architecture, and it is the piece most hobbyist implementations skip. A session agent can fake it with a loop; an always-on agent needs the real thing, with checkpointing on top so a crash mid-task resumes rather than restarts.

The spend-cap discipline

The budget node is the heart of the always-on pattern. A session agent that runs a hundred times a month is a rounding error; an always-on agent that wakes hourly is a recurring liability unless spend is governed structurally. spark-sleeper enforces the cap before every model call: the budget node reads the month's ledger, and if spent >= cap, the workflow pauses instead of spending. The ledger is month-keyed, so the cap resets naturally, and every spend is recorded per task so the user can see which agent activity cost what. That is the same unit-economics discipline the AI workflows library applies to every agent fleet — the difference is that an always-on agent makes it mandatory.

Cloud continuity, not device dependency

The second pillar is continuity. spark-sleeper runs on cloud compute — a small VM, a serverless container, or a cloud job — and your devices are only endpoints: they enqueue tasks and receive reports. The scheduler wakes the agent on a cron or trigger; the executor runs with checkpointing, so an hours-long task survives crashes and can resume from its last saved step. This is exactly the property Gemini Spark's marketing leads with — "works in the background 24/7, even if your phone and laptop are turned off" — implemented as an open workflow with your own keys. The same pattern appears across the AI workflows library for long-horizon agent tasks.

The bottom line

Gemini Spark's move to the $19.99 tier on July 25, 2026 made always-on personal agents a consumer product; spark-sleeper is the open pattern underneath it. Durable task queue, schedule and trigger wake-ups, cloud continuity with checkpointing, and a hard spend cap enforced before every call. The agent works while you sleep, and it cannot spend more than you decide. The patterns are in the AI workflows library; the always-on agent coverage is on latest AI news.

Visibility is part of the control

The cap should be visible to the user, not just enforced in code. Spark's consumer framing — "under your direction" — implies the user knows what the agent is doing and what it costs. spark-sleeper surfaces the ledger in the report, so the monthly summary shows what the agent did and what it spent. Visibility is what makes the cap a control instead of a surprise, and it is the same transparency the MCP directory applies to every governed tool surface.

Visibility is part of the control

The cap should be visible to the user, not just enforced in code. Spark's consumer framing — "under your direction" — implies the user knows what the agent is doing and what it costs. spark-sleeper surfaces the ledger in the report, so the monthly summary shows what the agent did and what it spent. Visibility is what makes the cap a control instead of a surprise, and it is the same transparency the MCP directory applies to every governed tool surface.

Frequently Asked Questions

What is spark-sleeper?

A LangGraph workflow that runs a personal AI agent 24/7 on cloud compute: it ingests tasks from any device, wakes on schedule or trigger, executes long-running work with a durable task queue and checkpointing, enforces a hard monthly spend cap, and reports results back to the user's channel.

Why build it now?

Gemini Spark dropped to the $19.99 AI Pro tier on July 25, 2026, proving always-on personal agents are a consumer product. spark-sleeper gives you the same pattern as an open workflow you control, with your own model keys and spend caps.

How does the agent run while my laptop is off?

The workflow runs on cloud compute (a small VM or serverless container). Your devices only enqueue tasks and receive reports; the agent itself never depends on a device being awake.

How are spend caps enforced?

A budget node tracks token and API spend per run and against a monthly cap before every model call. If the cap would be exceeded, the node pauses the agent and notifies you instead of spending.

What happens if a long task fails?

Every step checkpoints state to the task store. On failure, the workflow retries with backoff per the retry rules, and a failed task can be resumed from its last checkpoint rather than restarted.

Closing thoughts

The always-on agent era started with pricing, not models: Gemini Spark at $19.99 made 24/7 personal agents a commodity. spark-sleeper is the open workflow that gives you the same capability with your own keys, your own queue, and your own caps. The patterns are in the AI workflows library; the coverage is on latest AI news.