Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 110 additions & 1 deletion agents/langchain-deepagents-code/patch-managed-deepagents-code.py
Original file line number Diff line number Diff line change
Expand Up @@ -748,9 +748,102 @@ def _run_single_hook(command, event, payload_bytes) -> None:
del command, event, payload_bytes
'''

NON_INTERACTIVE_ERROR_MARKER = ''' except Exception as e:
logger.exception("Unexpected error during non-interactive execution")
console.print(
f"\\n[red]Unexpected error ({type(e).__name__}): "
f"{escape_markup(str(e))}[/red]"
)
return 1
'''

NON_INTERACTIVE_ERROR_PATCH = ''' except Exception:
_nemoclaw_report_non_interactive_error(thread_id, console)
return 1
'''

NON_INTERACTIVE_PATCH = r'''
import logging as _nemoclaw_logging
import os as _nemoclaw_os
import re as _nemoclaw_re
import sqlite3 as _nemoclaw_sqlite3

# NemoClaw-managed Deep Agents Code hardening v2.
_NEMOCLAW_PROVIDER_CAPACITY_ERROR = _nemoclaw_re.compile(
r"""\bAPIError\(["']ResourceExhausted:\s*
Worker\s+local\s+total\s+request\s+limit\s+reached\s*
\(\d+/\d+\)""",
_nemoclaw_re.IGNORECASE | _nemoclaw_re.VERBOSE,
)

_NEMOCLAW_MANAGED_STATE_DB = "/sandbox/.deepagents/.state/sessions.db"


def _nemoclaw_classify_persisted_error(thread_id):
"""Classify the observed provider-capacity error for one managed thread."""
if (
not isinstance(thread_id, str)
or not thread_id
or not _nemoclaw_os.path.isfile(_NEMOCLAW_MANAGED_STATE_DB)
):
return None
try:
conn = _nemoclaw_sqlite3.connect(_NEMOCLAW_MANAGED_STATE_DB, timeout=2)
conn.execute("PRAGMA query_only = ON")
try:
cursor = conn.execute(
"SELECT substr(value, 1, 4096) FROM writes "
"WHERE thread_id = ? AND channel = '__error__' "
"ORDER BY rowid DESC LIMIT 5",
(thread_id,),
)
for (value,) in cursor:
if not isinstance(value, (str, bytes)):
continue
text = (
value
if isinstance(value, str)
else value.decode("utf-8", errors="replace")
)
if _NEMOCLAW_PROVIDER_CAPACITY_ERROR.search(text):
return ("ResourceExhausted", "upstream_provider_capacity", "true")
finally:
conn.close()
except Exception:
# Diagnostics must not replace the original non-interactive exit result.
pass
return None


def _nemoclaw_report_non_interactive_error(thread_id, console):
"""Emit bounded diagnostics without logging the exception or checkpoint row."""
classified = _nemoclaw_classify_persisted_error(thread_id)
logger = _nemoclaw_logging.getLogger("nemoclaw.managed.non_interactive")
if classified:
error_class, category, retryable = classified
logger.warning(
"managed non-interactive error: error_class=%s category=%s "
"retryable=%s correlation_id=%s",
error_class,
category,
retryable,
thread_id,
)
console.print(
f"\n[red]Model request failed: {error_class} "
f"(correlation_id={thread_id})[/red]"
)
return
logger.warning(
"managed non-interactive error: error_class=unknown category=unknown "
"retryable=false correlation_id=%s",
thread_id,
)
console.print(
f"\n[red]Unexpected error (correlation_id={thread_id})[/red]"
)


_nemoclaw_original_run_non_interactive = run_non_interactive


Expand Down Expand Up @@ -1368,6 +1461,7 @@ def main() -> None:
("status", STATUS_PATCH),
("welcome", WELCOME_PATCH),
("server", SERVER_PATCH),
("non_interactive", NON_INTERACTIVE_PATCH),
):
if texts[name].count(patch.lstrip()) != 1:
raise RuntimeError(
Expand All @@ -1378,6 +1472,11 @@ def main() -> None:
"Managed package server override patch is incomplete in "
f"{paths['server']}"
)
if texts["non_interactive"].count(NON_INTERACTIVE_ERROR_PATCH) != 1:
raise RuntimeError(
"Managed package non-interactive error patch is incomplete in "
f"{paths['non_interactive']}"
)
return
if marker_states != {False} or helper_path.exists():
raise RuntimeError("Managed package patch is partial; refusing mixed source state")
Expand Down Expand Up @@ -1549,6 +1648,11 @@ def main() -> None:
"Expected one Deep Agents Code explicit MCP config marker in "
f"{paths['mcp_tools']}"
)
if texts["non_interactive"].count(NON_INTERACTIVE_ERROR_MARKER) != 1:
raise RuntimeError(
"Expected one Deep Agents Code non-interactive error marker in "
f"{paths['non_interactive']}"
)
transformed = dict(texts)
transformed["entrypoint"] = texts["entrypoint"].replace(
ENTRYPOINT_MARKER, ENTRYPOINT_PATCH, 1
Expand Down Expand Up @@ -1637,9 +1741,14 @@ def main() -> None:
transformed["hooks"] = _append_patch(
paths["hooks"], texts["hooks"], HOOKS_PATCH
)
transformed_non_interactive = texts["non_interactive"].replace(
NON_INTERACTIVE_ERROR_MARKER,
NON_INTERACTIVE_ERROR_PATCH,
1,
)
transformed["non_interactive"] = _append_patch(
paths["non_interactive"],
texts["non_interactive"],
transformed_non_interactive,
NON_INTERACTIVE_PATCH,
)

Expand Down
31 changes: 30 additions & 1 deletion test/helpers/langchain-deepagents-code-patch-fixture.ts
Original file line number Diff line number Diff line change
Expand Up @@ -429,16 +429,45 @@ def _run_single_hook(command, event, payload_bytes):
`
from __future__ import annotations

import logging
from types import SimpleNamespace

settings = SimpleNamespace(shell_allow_list=["bash"])
logger = logging.getLogger(__name__)


async def run_non_interactive(*args, **kwargs):
class _Console:
def print(self, message):
print(message)


def escape_markup(value):
return value


def generate_thread_id():
return "thread-1"


async def _run_non_interactive_impl(*args, **kwargs):
del args
return kwargs


async def run_non_interactive(*args, **kwargs):
console = _Console()
thread_id = generate_thread_id()
try:
return await _run_non_interactive_impl(*args, **kwargs)
except Exception as e:
logger.exception("Unexpected error during non-interactive execution")
console.print(
f"\\n[red]Unexpected error ({type(e).__name__}): "
f"{escape_markup(str(e))}[/red]"
)
return 1


async def _run_startup_command(command, console, *, quiet):
del console, quiet
return command
Expand Down
187 changes: 187 additions & 0 deletions test/non-interactive-error-classification.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,187 @@
// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

import { spawnSync } from "node:child_process";
import { afterEach, describe, expect, it } from "vitest";
import {
cleanupPackageFixtures,
createPackageFixture,
patchFixture,
} from "./helpers/langchain-deepagents-code-patch-fixture";
Comment thread
coderabbitai[bot] marked this conversation as resolved.

afterEach(cleanupPackageFixtures);

function runPatchedNonInteractive(driver: string) {
const tempDir = createPackageFixture();
patchFixture(tempDir);

return spawnSync("python3", ["-c", driver], {
encoding: "utf8",
env: {
PATH: process.env.PATH,
PYTHONPATH: tempDir,
},
timeout: 10_000,
});
}

const runtimePreamble = `
import asyncio
import logging
import os
import sqlite3
import tempfile

from deepagents_code.client import non_interactive as target

logging.basicConfig(level=logging.WARNING, format="%(message)s")


class RemoteException(Exception):
pass


async def fail(*args, **kwargs):
del args, kwargs
raise RemoteException("remote failure token=runtime-secret")


target._run_non_interactive_impl = fail
`;

describe("managed non-interactive error reporting", () => {
it("reports the provider-capacity error persisted for the failing thread (#7415)", () => {
const result = runPatchedNonInteractive(`
${runtimePreamble}
handle, db_path = tempfile.mkstemp()
os.close(handle)
target._NEMOCLAW_MANAGED_STATE_DB = db_path
target.generate_thread_id = lambda: "thread-current"

connection = sqlite3.connect(db_path)
connection.execute("CREATE TABLE writes (thread_id TEXT, channel TEXT, value BLOB)")
connection.execute(
"INSERT INTO writes (thread_id, channel, value) VALUES (?, ?, ?)",
(
"thread-current",
"__error__",
sqlite3.Binary(
b"APIError('ResourceExhausted: Worker local total request limit reached "
b"(32/32) token=checkpoint-secret request=private model=private-model')"
),
),
)
connection.commit()
connection.close()

exit_code = asyncio.run(target.run_non_interactive("task"))
assert exit_code == 1
os.unlink(db_path)
`);

expect(result.status).toBe(0);
expect(result.stderr).toContain(
"error_class=ResourceExhausted category=upstream_provider_capacity retryable=true correlation_id=thread-current",
);
expect(result.stdout).toContain(
"Model request failed: ResourceExhausted (correlation_id=thread-current)",
);
expect(`${result.stdout}\n${result.stderr}`).not.toMatch(
/runtime-secret|checkpoint-secret|request=private|private-model/,
);
});

it("does not use a provider error from another thread (#7415)", () => {
const result = runPatchedNonInteractive(`
${runtimePreamble}
handle, db_path = tempfile.mkstemp()
os.close(handle)
target._NEMOCLAW_MANAGED_STATE_DB = db_path
target.generate_thread_id = lambda: "thread-current"

connection = sqlite3.connect(db_path)
connection.execute("CREATE TABLE writes (thread_id TEXT, channel TEXT, value BLOB)")
connection.execute(
"INSERT INTO writes (thread_id, channel, value) VALUES (?, ?, ?)",
(
"thread-other",
"__error__",
"APIError('ResourceExhausted: Worker local total request limit reached (32/32)')",
),
)
connection.execute(
"INSERT INTO writes (thread_id, channel, value) VALUES (?, ?, ?)",
("thread-current", "__error__", "APIError('MCP connection refused')"),
)
connection.commit()
connection.close()

exit_code = asyncio.run(target.run_non_interactive("task"))
assert exit_code == 1
os.unlink(db_path)
`);

expect(result.status).toBe(0);
expect(result.stderr).toContain(
"error_class=unknown category=unknown retryable=false correlation_id=thread-current",
);
expect(result.stdout).toContain("Unexpected error (correlation_id=thread-current)");
expect(`${result.stdout}\n${result.stderr}`).not.toContain("upstream_provider_capacity");
});

it("keeps MCP, gateway, and policy errors unknown (#7415)", () => {
const result = runPatchedNonInteractive(`
${runtimePreamble}
handle, db_path = tempfile.mkstemp()
os.close(handle)
target._NEMOCLAW_MANAGED_STATE_DB = db_path

connection = sqlite3.connect(db_path)
connection.execute("CREATE TABLE writes (thread_id TEXT, channel TEXT, value BLOB)")
errors = {
"thread-mcp": "APIError('MCP connection refused')",
"thread-gateway": "APIError('gateway timeout')",
"thread-policy": "APIError('policy returned 429')",
}
connection.executemany(
"INSERT INTO writes (thread_id, channel, value) VALUES (?, '__error__', ?)",
errors.items(),
)
connection.commit()
connection.close()

for thread_id in errors:
target.generate_thread_id = lambda current=thread_id: current
exit_code = asyncio.run(target.run_non_interactive("task"))
assert exit_code == 1

os.unlink(db_path)
`);

expect(result.status).toBe(0);
expect(result.stderr.match(/error_class=unknown/g)).toHaveLength(3);
expect(result.stdout.match(/Unexpected error/g)).toHaveLength(3);
expect(`${result.stdout}\n${result.stderr}`).not.toContain("upstream_provider_capacity");
});

it("sanitizes unknown failures when checkpoint diagnostics are unavailable (#7415)", () => {
const result = runPatchedNonInteractive(`
${runtimePreamble}
handle, db_path = tempfile.mkstemp()
os.close(handle)
target._NEMOCLAW_MANAGED_STATE_DB = db_path
target.generate_thread_id = lambda: "thread-no-diagnostics"

exit_code = asyncio.run(target.run_non_interactive("task"))
assert exit_code == 1
os.unlink(db_path)
`);

expect(result.status).toBe(0);
expect(result.stderr).toContain(
"error_class=unknown category=unknown retryable=false correlation_id=thread-no-diagnostics",
);
expect(result.stdout).toContain("Unexpected error (correlation_id=thread-no-diagnostics)");
expect(`${result.stdout}\n${result.stderr}`).not.toContain("runtime-secret");
});
});
Loading