Agents That Plateau: Inside DigitalOcean's Learning and Data Services

DigitalOcean announced Agent Learning and Data Services on October 12, 2026, ahead of its Open Intelligence Summit, and the framing is the interesting part: Inference Engine so agents could think, Managed Agents so they could act, and now Learning and Data Services so they can improve. The company's own diagnosis of the problem is the most quotable thing in the release — once an agent is deployed, it plateaus, and the tools watching it can tell you the system is up but not why the agent called the wrong tool. Here is what shipped, what is still behind a preview, and which four pieces of the loop you should own regardless of vendor.

Getting better with every run is what the announcement is really selling

What was announced

Agent Learning Services is three products that work as one loop. Insights records what an agent does; Signals flags where it falls short; Evaluations and Simulations test a change before customers see it. The evidence those three produce is what a team uses to optimise an agent, rather than guessing at a prompt edit. Agent Data Services is the layer underneath: one managed place for agents to read the context they need and write back what they learn. Vinay Kumar, chief product and technology officer, describes the shift as the question changing from whether a team can launch an agent to whether that agent gets better with every run. The services are positioned as running on the customer's own data, on an open cloud, rather than inside a single model vendor's platform.

# Insights, at its core, is a trace you own. Write it with redaction, not after.
import json, re, hashlib, time

SECRET = re.compile(r"(sk-[A-Za-z0-9]{8,}|Bearer\s+[A-Za-z0-9._-]+|\b\d{13,19}\b)")
FIELDS = {"email", "phone", "card", "address", "token", "api_key"}

def redact(value, path=""):
    if isinstance(value, dict):
        return {k: redact(v, k) for k, v in value.items()}
    if isinstance(value, list):
        return [redact(v, path) for v in value]
    if isinstance(value, str):
        if path in FIELDS:
            return "hash:" + hashlib.sha256(value.encode()).hexdigest()[:16]
        return SECRET.sub("[redacted]", value)
    return value

def trace(session_id, step, tool, args, result, tokens=None):
    return {
        "ts": round(time.time(), 3),
        "session_id": session_id,
        "step": step,
        "tool": tool,
        "args": redact(args, tool),
        "result_digest": hashlib.sha256(json.dumps(result, sort_keys=True).encode()).hexdigest()[:12],
        "result_chars": len(json.dumps(result)),
        "tokens": tokens,
    }

print(json.dumps(trace("s-1", 1, "search", {"q": "renewal date", "api_key": "sk-abcdef12345678"},
                       {"hits": 3}), indent=2))

# DigitalOcean describes this as assembling logs and agent sessions with
# configurable redaction controls. Configurable means decided before you write.

The plateau problem, described accurately for once

The release's own problem statement is unusually honest. Once an agent is deployed, it plateaus, because every piece of the job lives somewhere else. The record of what the agent did sits in a monitoring tool that shows whether systems are up, not why the agent called the wrong tool or lost a user. A fix is usually a prompt tweak with no way to test it, so the first person to notice a regression is a customer. Improving the model means exporting production data to a training vendor and wiring together four or five tools, so it happens once a quarter or not at all. Almost every team running agents in production in 2026 would recognise at least two of those four sentences, which is why the framing lands.

# Signals: flag the sessions that matter. Two cheap detectors catch most of them.
import json
from collections import Counter

def load(path):
    return [json.loads(line) for line in open(path) if line.strip()]

def looped(session, repeat_threshold=3, window=6):
    keys = [(s["tool"], json.dumps(s["args"], sort_keys=True)) for s in session]
    recent = keys[-window:]
    for key, count in Counter(recent).items():
        if count >= repeat_threshold:
            return {"kind": "loop", "tool": key[0], "repeats": count}
    return None

def stalled(session, token_budget=40000):
    spent = sum(s.get("tokens") or 0 for s in session)
    if spent > token_budget:
        return {"kind": "budget", "tokens": spent}
    if session and session[-1]["tool"] == "ask_user":
        return {"kind": "lost_user", "step": session[-1]["step"]}
    return None

sessions = {}
for row in load("traces.jsonl"):
    sessions.setdefault(row["session_id"], []).append(row)

flagged = {sid: (looped(s) or stalled(s)) for sid, s in sessions.items()}
for sid, signal in flagged.items():
    if signal:
        print(sid, signal)

# Each flagged session becomes a candidate case for Evaluations and Simulations.

Sessions, signals and evaluations wired into one loop

Insights, Signals, Evaluations: the loop itself

The loop is familiar to anyone who has built an evaluation harness by hand, with the ordering made explicit. Insights records every step an agent takes. Signals flags the sessions that matter — where the agent stalled, looped, lost the user or failed. That evidence becomes a proposed change, and Evaluations and Simulations test it ahead of real users using realistic conversations built from the flagged sessions. Each change then feeds the next turn of the loop. DigitalOcean says recursive self-improvement is available today for models through the enhanced Inference Router, which picks the best-scoring model for each task, optimised for cost, quality or the team's own evaluations, and re-runs those evaluations as new models ship. Extending that automated loop to agent workflows is described as planned, augmenting rather than replacing human approvals.

# Evaluations: run a candidate change against the cases you harvested.
import json

def runner(prompt_version, case):
    # Stand-in for your agent. Swap in a real invocation of the candidate.
    expected = case["expect"]["tool"]
    known_good = {"v1": expected, "v2": expected if case["difficulty"] < 3 else "ask_user"}
    return {"tool": known_good.get(prompt_version, "ask_user"), "tokens": 1200}

def evaluate(cases, versions):
    table = {}
    for version in versions:
        passed = 0
        for case in cases:
            out = runner(version, case)
            if out["tool"] == case["expect"]["tool"]:
                passed += 1
        table[version] = round(100 * passed / len(cases), 1)
    return table

CASES = [
    {"id": "c1", "difficulty": 1, "expect": {"tool": "search"}},
    {"id": "c2", "difficulty": 2, "expect": {"tool": "update_ticket"}},
    {"id": "c3", "difficulty": 4, "expect": {"tool": "escalate"}},
]
print(evaluate(CASES, ["v1", "v2"]))

# The value is not the score. It is that the score is computed before a customer
# discovers the regression, from sessions that were already flagged in production.

The data layer underneath

Agent Data Services is the part that decides whether the loop works at all, because an evaluation is only as good as the context you can replay. In DigitalOcean's description, every piece of company data becomes agent-ready as it arrives: video is transcribed and described, documents are read and indexed, logs and agent sessions are assembled with configurable redaction controls for sensitive details, and all of it is searchable by meaning in a single call. It lands in one catalog that is logically isolated to that customer, with governance designed so that only authorised people and agents can reach each piece. Signals writes sessions into it, and Optimizer draws training and test data from it only where the customer expressly opts in. The same layer is offered for plain search and analytics across video, audio, documents and structured data.

# Optimizer data is opt-in per customer. Encode the gate, do not document it.
import json
from datetime import date

CUSTOMERS = [
    {"tenant": "acme", "training_opt_in": True, "signed": "2026-09-01"},
    {"tenant": "northwind", "training_opt_in": False, "signed": None},
]

OPT_IN_VERSION = "2026-10"

def exportable(record, customer):
    if not customer["training_opt_in"]:
        return False, "no training opt-in on file"
    if record.get("contains_third_party_data"):
        return False, "session contains third-party data"
    if record["tenant"] != customer["tenant"]:
        return False, "tenant mismatch"
    return True, "ok"

def build_dataset(records, customers):
    index = {c["tenant"]: c for c in customers}
    dataset, skipped = [], []
    for rec in records:
        ok, why = exportable(rec, index[rec["tenant"]])
        (dataset if ok else skipped).append((rec["id"], why))
    return dataset, skipped

print(build_dataset([{"id": "r1", "tenant": "acme", "contains_third_party_data": False},
                     {"id": "r2", "tenant": "northwind", "contains_third_party_data": False}],
                    CUSTOMERS))

# "Only where the customer expressly opts in" is a query condition, not a policy page.

Redaction, tracing and reuse are decided at write time

What is available, and what is not

Agent Learning Services, meaning Insights, Signals, and Evaluations and Simulations, are available now in public preview. The Inference Router optimizer and Agent Data Services are in private preview, with access requested through a separate form. There is no published pricing for either in the announcement, and no independent benchmarks of the loop improving an agent in production. One customer is quoted by name: Amplitude, whose chief executive Spenser Skates describes choosing the model for each task based on the results it delivers rather than cost alone, with Managed Agents handling the infrastructure as it scales. Treat the availability split as the practical fact: the observability and evaluation pieces you can try today, the data layer and the router optimiser you cannot yet.

# Tenant isolation: every read path should be filtered, including the debug path.
import json

RECORDS = [
    {"id": "r1", "tenant": "acme", "kind": "session"},
    {"id": "r2", "tenant": "acme", "kind": "document"},
    {"id": "r3", "tenant": "northwind", "kind": "session"},
]

ALLOWED_KINDS = {"acme": {"session", "document"}, "northwind": {"session"}}

def search(records, tenant, query="", kinds=None, actor="agent"):
    kinds = kinds or ALLOWED_KINDS[tenant]
    if kinds - ALLOWED_KINDS[tenant]:
        raise PermissionError("actor " + actor + " requested kinds outside the tenant grant")
    return [r for r in records
            if r["tenant"] == tenant and r["kind"] in kinds and query.lower() in json.dumps(r).lower()]

print(search(RECORDS, "acme", query=""))
print(search(RECORDS, "northwind", query=""))

# The same function should back your console, your API and your agent tool calls.
# An unfiltered path is a data leak with a friendly interface.

The four pieces to own either way

Whether or not you buy any of this, four components make up the loop and all four are things you can build against interfaces you already have. A durable trace written at call time, with redaction decided in the writer rather than in a downstream job, because the data you failed to redact is the data you cannot un-store. A signal detector for the two cheap failures, loops and budget blowouts, since those are detectable with counters rather than models. A replayable evaluation set harvested from flagged sessions, so a prompt change has a score before it has users. And an explicit opt-in gate on any path that turns production sessions into training or test data, enforced in code rather than described in a policy page.