langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 01:35:28 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

feat(code): non-interactive rubric grading flags (#4305)

Browse Source 
Wires `deepagents-code`'s headless (`-n`) path to the SDK's
`RubricMiddleware` so a run can self-evaluate against caller-supplied
acceptance criteria and loop until satisfied. This is the
non-interactive vertical slice of the rubric/"acceptance criteria"
feature; the interactive TUI `/rubric` command + status indicator are
intentionally left as a follow-up.

New flags (all `-n`/piped-stdin only):
* `--rubric TEXT|@PATH` accepts literal criteria, or reads criteria from
`@path`. `@path` may be absolute, relative to the `dcode` process
working directory, or use `~` for the home directory.
* `--rubric-model MODEL` defaults to the main agent model.
* `--rubric-max-iterations N` controls grader iterations per attempt.

The rubric text is passed as graph `rubric` state; grader model +
max-iterations flow to the server subprocess via `ServerConfig`. Grader
lifecycle events are surfaced by subscribing to the `custom` stream and
rendering `rubric_evaluation_start/end` (⏳ grading / ✓ satisfied / ↻
needs revision / ⚠ max-iterations / failed), and the headless header
shows `Rubric: active`.

## Test Plan
- [ ] `dcode -n "implement X" --rubric "tests pass; minimal diff"`
against an SDK that ships `RubricMiddleware` loops the agent and renders
grading events
- [ ] `--rubric @rubric.md` resolves file contents from a relative path;
absolute paths and `~`-expanded paths are also supported
- [ ] `--rubric` without `-n` exits 2

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
2026-06-29 00:50:16 -04:00
committed by GitHub
parent 90ebb1d68c
commit 1bcb112ee7
11 changed files with 602 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/deepagents_code/_server_config.py
+39 -3
View File
@@ -12,7 +12,6 @@ with `from_env()`.
from __future__ import annotations
import json
import logging
import os
from dataclasses import dataclass
from pathlib import Path
@@ -24,8 +23,6 @@ from deepagents_code._env_vars import SERVER_ENV_PREFIX
if TYPE_CHECKING:
from deepagents_code.project_utils import ProjectContext
logger = logging.getLogger(__name__)
def _read_env_bool(suffix: str, *, default: bool = False) -> bool:
"""Read a `DEEPAGENTS_CODE_SERVER_*` boolean from the environment.
@@ -83,6 +80,25 @@ def _read_env_str(suffix: str) -> str | None:
return os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
def _read_env_int(suffix: str, *, default: int) -> int:
"""Read a `DEEPAGENTS_CODE_SERVER_*` integer from the environment.
Args:
suffix: Variable name suffix after the `DEEPAGENTS_CODE_SERVER_` prefix.
default: Value when the variable is absent or malformed.
Returns:
Parsed integer, or the default when parsing fails.
"""
raw = os.environ.get(f"{SERVER_ENV_PREFIX}{suffix}")
if raw is None:
return default
try:
return int(raw)
except ValueError:
return default
def _read_env_optional_bool(suffix: str) -> bool | None:
"""Read a tri-state `DEEPAGENTS_CODE_SERVER_*` boolean (`True` / `False` / `None`).
@@ -242,6 +258,16 @@ class ServerConfig:
`interpreter_ptc="all"` is paired with non-`auto_approve` mode.
"""
rubric_model: str | None = None
"""Grader model spec for `RubricMiddleware` (e.g. `'anthropic:...'`).
`None` reuses the main agent model.
"""
rubric_max_iterations: int = 3
"""Grader iterations per rubric attempt before the agent stops with
`'max_iterations_reached'`."""
sandbox_type: str | None = None
"""Sandbox backend identifier (e.g. `'daytona'`); `None` runs tools on the
host. `'none'` is normalized to `None` in `__post_init__`."""
@@ -329,6 +355,8 @@ class ServerConfig:
"INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE": str(
self.interpreter_ptc_acknowledge_unsafe
).lower(),
"RUBRIC_MODEL": self.rubric_model,
"RUBRIC_MAX_ITERATIONS": str(self.rubric_max_iterations),
"SANDBOX_TYPE": self.sandbox_type,
"SANDBOX_ID": self.sandbox_id,
"SANDBOX_SNAPSHOT_NAME": self.sandbox_snapshot_name,
@@ -377,6 +405,8 @@ class ServerConfig:
interpreter_ptc_acknowledge_unsafe=_read_env_bool(
"INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE"
),
rubric_model=_read_env_str("RUBRIC_MODEL"),
rubric_max_iterations=_read_env_int("RUBRIC_MAX_ITERATIONS", default=3),
sandbox_type=_read_env_str("SANDBOX_TYPE"),
sandbox_id=_read_env_str("SANDBOX_ID"),
sandbox_snapshot_name=_read_env_str("SANDBOX_SNAPSHOT_NAME") or None,
@@ -412,6 +442,8 @@ class ServerConfig:
enable_interpreter: bool | None = None,
interpreter_ptc: str | list[str] | None = None,
interpreter_ptc_acknowledge_unsafe: bool = False,
rubric_model: str | None = None,
rubric_max_iterations: int = 3,
mcp_config_path: str | None,
no_mcp: bool,
trust_project_mcp: bool | None,
@@ -445,6 +477,8 @@ class ServerConfig:
interpreter_ptc: Override for `settings.interpreter_ptc`.
interpreter_ptc_acknowledge_unsafe: Mirror of
`settings.interpreter_ptc_acknowledge_unsafe`.
rubric_model: Grader model spec; `None` reuses the main model.
rubric_max_iterations: Grader iterations per rubric attempt.
mcp_config_path: Path to MCP config.
no_mcp: Disable MCP.
trust_project_mcp: Trust project MCP servers.
@@ -472,6 +506,8 @@ class ServerConfig:
enable_interpreter=resolved_enable_interpreter,
interpreter_ptc=interpreter_ptc,
interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe,
rubric_model=rubric_model,
rubric_max_iterations=rubric_max_iterations,
sandbox_type=sandbox_type,
sandbox_id=sandbox_id,
sandbox_snapshot_name=sandbox_snapshot_name,
libs/code/deepagents_code/agent.py
+26 -1
View File
@@ -9,13 +9,14 @@ import re
import shutil
import tempfile
import tomllib
import warnings
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from deepagents import create_deep_agent
from deepagents.backends import CompositeBackend, LocalShellBackend
from deepagents.backends.filesystem import FilesystemBackend
from deepagents.middleware import MemoryMiddleware, SkillsMiddleware
from deepagents.middleware import MemoryMiddleware, RubricMiddleware, SkillsMiddleware
# Backwards-compat flag: SDKs before 0.5.4 accept only `list[str]` for
# `SkillsMiddleware.sources`; newer SDKs expose the `SkillSource` alias
@@ -1249,6 +1250,8 @@ def create_cli_agent(
enable_skills: bool = True,
enable_shell: bool = True,
enable_interpreter: bool = False,
rubric_model: str | BaseChatModel | None = None,
rubric_max_iterations: int = 3,
checkpointer: BaseCheckpointSaver | None = None,
mcp_server_info: list[MCPServerInfo] | None = None,
cwd: str | Path | None = None,
@@ -1325,6 +1328,13 @@ def create_cli_agent(
`interpreter_ptc_acknowledge_unsafe=True`.
Requires the core `langchain-quickjs` dependency.
rubric_model: Grader model for `RubricMiddleware`.
A `'provider:model'` string or `BaseChatModel`.
When `None`, the main `model` is reused.
rubric_max_iterations: Grader iterations per rubric attempt before the
agent terminates with `'max_iterations_reached'`.
checkpointer: Optional checkpointer for session persistence.
When `None`, the graph is compiled without a checkpointer.
mcp_server_info: MCP server metadata to surface in the system prompt.
@@ -1707,6 +1717,21 @@ def create_cli_agent(
create_summarization_tool_middleware(model, composite_backend)
)
# Rubric-driven self-evaluation. The middleware is a no-op until a
# `rubric` is supplied on invocation state, so installing it is safe.
with warnings.catch_warnings():
warnings.filterwarnings(
"ignore",
message="The middleware `RubricMiddleware` is in beta",
category=Warning,
)
agent_middleware.append(
RubricMiddleware(
model=rubric_model if rubric_model is not None else model,
max_iterations=rubric_max_iterations,
)
)
# Create the agent
all_subagents: list[SubAgent | CompiledSubAgent | AsyncSubAgent] = [
*custom_subagents,
libs/code/deepagents_code/main.py
+99
View File
@@ -625,6 +625,48 @@ def _warn_if_interpreter_disabled_by_sandbox(args: argparse.Namespace) -> None:
)
def _resolve_rubric_text(rubric: str | None) -> str | None:
"""Resolve the rubric from `--rubric` into one string.
`--rubric` accepts literal text, or `@path` to read a file. File paths
may be absolute, relative to the `dcode` process working directory, or
`~`-expanded home paths.
Args:
rubric: Value of `--rubric` (literal text or `@path`), or `None`.
Returns:
The resolved rubric text, or `None` when the flag was not supplied.
Raises:
ValueError: If the rubric is empty, or a referenced file is missing,
unreadable, or empty.
"""
if rubric is None:
return None
# An `@`-prefixed value is always read as a file path. The path may be
# absolute, relative to the `dcode` process working directory, or `~`-based.
# There is no way to pass a literal rubric that begins with `@` (put such
# text in a file).
if rubric.startswith("@"):
path = rubric[1:]
try:
text = Path(path).expanduser().read_text(encoding="utf-8")
except OSError as exc:
msg = f"Could not read rubric file {path!r}: {exc}."
raise ValueError(msg) from exc
if not text.strip():
msg = f"Rubric file {path!r} is empty."
raise ValueError(msg)
return text.strip()
if not rubric.strip():
msg = "--rubric must not be empty."
raise ValueError(msg)
return rubric.strip()
def _warn_if_interpreter_tools_without_interpreter(
args: argparse.Namespace, *, enable_interpreter: bool
) -> None:
@@ -1510,6 +1552,31 @@ def parse_args() -> argparse.Namespace:
"use exit code 124 on expiry. Requires -n or piped stdin.",
)
parser.add_argument(
"--rubric",
dest="rubric",
metavar="TEXT|@PATH",
help="Acceptance criteria the agent self-evaluates against, looping "
"until satisfied. Accepts literal text or '@path' to read a file "
"(relative to the current working directory; '~' supported). "
"Requires -n or piped stdin.",
)
parser.add_argument(
"--rubric-model",
dest="rubric_model",
metavar="MODEL",
help="Model the rubric grader uses (e.g. anthropic:claude-sonnet-4-6). "
"Defaults to the main agent model.",
)
parser.add_argument(
"--rubric-max-iterations",
dest="rubric_max_iterations",
type=positive_int,
metavar="N",
help="Grader iterations per rubric attempt before stopping (must be "
">= 1, default 3).",
)
parser.add_argument(
"--stdin",
action="store_true",
@@ -2626,6 +2693,25 @@ def cli_main() -> None:
)
sys.exit(2)
rubric_set = any(
getattr(args, attr, None) is not None
for attr in (
"rubric",
"rubric_model",
"rubric_max_iterations",
)
)
if rubric_set and not args.non_interactive_message:
from rich.console import Console as _Console
_Console(stderr=True).print(
"[bold red]Error:[/bold red] --rubric/--rubric-model/"
"--rubric-max-iterations require "
"--non-interactive (-n) or piped stdin\n"
" dcode -n 'implement X' --rubric 'tests pass; minimal diff'"
)
sys.exit(2)
if (args.quiet or args.no_stream) and not args.non_interactive_message:
# Print to stderr (not the module-level stdout console) and exit
# with code 2 to match the POSIX convention for usage errors, as
@@ -3289,6 +3375,14 @@ def cli_main() -> None:
)
_warn_if_interpreter_disabled_by_sandbox(args)
try:
rubric_text = _resolve_rubric_text(getattr(args, "rubric", None))
except ValueError as exc:
from rich.console import Console as _Console
_Console(stderr=True).print(f"[bold red]Error:[/bold red] {exc}")
sys.exit(2)
timeout = getattr(args, "timeout", None)
try:
exit_code = asyncio.run(
@@ -3313,6 +3407,11 @@ def cli_main() -> None:
enable_interpreter=enable_interpreter,
interpreter_ptc=interpreter_ptc,
max_turns=getattr(args, "max_turns", None),
rubric=rubric_text,
rubric_model=getattr(args, "rubric_model", None),
rubric_max_iterations=getattr(
args, "rubric_max_iterations", None
),
),
timeout=timeout,
)
libs/code/deepagents_code/non_interactive.py
+94 -2
View File
@@ -481,6 +481,66 @@ def _process_message_chunk(
state.spinner.start()
def _process_rubric_event(
data: dict[str, Any],
state: StreamState,
console: Console,
) -> None:
"""Render a `RubricMiddleware` lifecycle event from the custom stream.
`RubricMiddleware` emits `rubric_evaluation_start` / `rubric_evaluation_end`
dicts via `runtime.stream_writer`. Non-rubric custom payloads are ignored.
Args:
data: The custom-stream payload dict.
state: Shared stream state (used to pause the spinner).
console: Rich console for status output (stderr in `--quiet` mode).
"""
event_type = data.get("type")
if event_type not in {"rubric_evaluation_start", "rubric_evaluation_end"}:
return
if state.spinner:
state.spinner.stop()
if event_type == "rubric_evaluation_start":
iteration = data.get("iteration", 0)
console.print(
f"[dim]⏳ Grading against rubric (iteration {iteration + 1})…[/dim]",
highlight=False,
)
if state.spinner:
state.spinner.start()
return
result = data.get("result")
explanation = (data.get("explanation") or "").strip()
if result == "satisfied":
console.print("[green]✓ Rubric satisfied[/green]", highlight=False)
elif result == "needs_revision":
suffix = f": {escape_markup(explanation)}" if explanation else ""
console.print(
f"[yellow]↻ Rubric needs revision{suffix}[/yellow]", highlight=False
)
for criterion in data.get("criteria", []):
if isinstance(criterion, dict) and not criterion.get("passed", True):
name = escape_markup(str(criterion.get("name", "criterion")))
gap = escape_markup(str(criterion.get("gap", "")).strip())
detail = f" — {gap}" if gap else ""
console.print(f"[yellow] ✗ {name}{detail}[/yellow]", highlight=False)
elif result == "max_iterations_reached":
console.print(
"[yellow]⚠ Rubric not satisfied (max iterations reached)[/yellow]",
highlight=False,
)
elif result == "failed":
suffix = f": {escape_markup(explanation)}" if explanation else ""
console.print(f"[red]⚠ Rubric grader failed{suffix}[/red]", highlight=False)
if state.spinner:
state.spinner.start()
def _process_stream_chunk(
chunk: object,
state: StreamState,
@@ -515,6 +575,8 @@ def _process_stream_chunk(
if stream_mode == "updates" and isinstance(data, dict) and "__interrupt__" in data:
_process_interrupts(cast("dict[str, list[Interrupt]]", data), state, console)
elif stream_mode == "custom" and isinstance(data, dict):
_process_rubric_event(cast("dict[str, Any]", data), state, console)
elif stream_mode == "messages":
_process_message_chunk(
cast("tuple[AIMessage | ToolMessage, dict[str, str]]", data),
@@ -676,7 +738,7 @@ async def _stream_agent(
try:
async for chunk in agent.astream(
stream_input,
stream_mode=["messages", "updates"],
stream_mode=["messages", "updates", "custom"],
subgraphs=True,
config=config,
context=context,
@@ -700,6 +762,7 @@ async def _run_agent_loop(
message_kwargs: dict[str, Any] | None = None,
thread_url_lookup: ThreadUrlLookupState | None = None,
max_turns: int | None = None,
rubric: str | None = None,
) -> None:
"""Run the agent and handle HITL interrupts until the task completes.
@@ -726,6 +789,10 @@ async def _run_agent_loop(
HITL resumes).
When `None`, falls back to `_MAX_HITL_ITERATIONS`.
rubric: Acceptance criteria supplied to `RubricMiddleware` via the
graph's `rubric` state field.
`None` leaves it unset (no grading).
Raises:
HITLIterationLimitError: If the effective turn limit is exceeded.
@@ -736,6 +803,8 @@ async def _run_agent_loop(
if message_kwargs:
user_msg.update(message_kwargs)
stream_input: dict[str, Any] | Command = {"messages": [user_msg]}
if rubric is not None:
stream_input["rubric"] = rubric
thread_id = config.get("configurable", {}).get("thread_id", "")
# An empty or missing thread ID carries no session identity, so leave it
@@ -813,6 +882,7 @@ def _build_non_interactive_header(
thread_id: str,
*,
include_thread_link: bool = False,
rubric_active: bool = False,
) -> Text:
"""Build the non-interactive mode header with model, agent, and thread info.
@@ -824,6 +894,8 @@ def _build_non_interactive_header(
thread_id: Thread identifier.
include_thread_link: Whether to resolve and render a LangSmith link for
the thread ID.
rubric_active: Whether a rubric is active for this run; when `True`,
appends a `Rubric: active` marker so the behavior change is visible.
Returns:
Rich Text object with the formatted header line.
@@ -851,6 +923,9 @@ def _build_non_interactive_header(
else:
parts.append((f"Thread: {thread_id}", "dim"))
if rubric_active:
parts.extend([(" | ", "dim"), ("Rubric: active", "dim")])
return Text.assemble(*parts)
@@ -950,6 +1025,9 @@ async def run_non_interactive(
interpreter_ptc: str | list[str] | None = None,
interpreter_ptc_acknowledge_unsafe: bool = False,
max_turns: int | None = None,
rubric: str | None = None,
rubric_model: str | None = None,
rubric_max_iterations: int | None = None,
) -> int:
"""Run a single task non-interactively and exit.
@@ -1013,6 +1091,13 @@ async def run_non_interactive(
`interpreter_ptc="all"` outside of `auto_approve`.
max_turns: Optional cap on total agentic turns. When `None`, the
internal safety default applies.
rubric: Acceptance criteria for `RubricMiddleware`. When provided, the
agent self-evaluates against it and loops until satisfied.
`None` disables rubric grading.
rubric_model: Grader model spec; `None` reuses the main model.
rubric_max_iterations: Grader iterations per rubric attempt; `None`
uses the middleware default.
Returns:
Exit code: 0 for success, 1 for error, 124 when the `--max-turns`
@@ -1127,7 +1212,9 @@ async def run_non_interactive(
if not quiet:
thread_url_lookup = _start_langsmith_thread_url_lookup(thread_id)
console.print(Text("Running task non-interactively...", style="dim"))
header = _build_non_interactive_header(assistant_id, thread_id)
header = _build_non_interactive_header(
assistant_id, thread_id, rubric_active=rubric is not None
)
console.print(header)
import asyncio
@@ -1187,6 +1274,10 @@ async def run_non_interactive(
enable_interpreter=enable_interpreter,
interpreter_ptc=interpreter_ptc,
interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe,
rubric_model=rubric_model,
rubric_max_iterations=(
rubric_max_iterations if rubric_max_iterations is not None else 3
),
mcp_config_path=mcp_config_path,
no_mcp=no_mcp,
trust_project_mcp=trust_project_mcp,
@@ -1222,6 +1313,7 @@ async def run_non_interactive(
message_kwargs=message_kwargs,
thread_url_lookup=thread_url_lookup,
max_turns=max_turns,
rubric=rubric,
)
except KeyboardInterrupt:
libs/code/deepagents_code/server_graph.py
+2
View File
@@ -255,6 +255,8 @@ async def _make_graph() -> Any: # noqa: ANN401
enable_skills=config.enable_skills,
enable_shell=config.enable_shell,
enable_interpreter=config.enable_interpreter,
rubric_model=config.rubric_model,
rubric_max_iterations=config.rubric_max_iterations,
mcp_server_info=mcp_server_info,
cwd=project_context.user_cwd if project_context is not None else config.cwd,
project_context=project_context,
libs/code/deepagents_code/server_manager.py
+12
View File
@@ -285,6 +285,8 @@ async def start_server_and_get_agent(
enable_interpreter: bool | None = None,
interpreter_ptc: str | list[str] | None = None,
interpreter_ptc_acknowledge_unsafe: bool = False,
rubric_model: str | None = None,
rubric_max_iterations: int = 3,
mcp_config_path: str | None = None,
no_mcp: bool = False,
trust_project_mcp: bool | None = None,
@@ -312,6 +314,8 @@ async def start_server_and_get_agent(
interpreter_ptc: Override for `settings.interpreter_ptc` (PTC allowlist).
interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for
`interpreter_ptc="all"` outside of `auto_approve`.
rubric_model: Grader model spec; `None` reuses the main model.
rubric_max_iterations: Grader iterations per rubric attempt.
mcp_config_path: Path to MCP config.
no_mcp: Disable MCP.
trust_project_mcp: Trust project MCP servers.
@@ -358,6 +362,8 @@ async def start_server_and_get_agent(
enable_interpreter=enable_interpreter,
interpreter_ptc=interpreter_ptc,
interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe,
rubric_model=rubric_model,
rubric_max_iterations=rubric_max_iterations,
mcp_config_path=mcp_config_path,
no_mcp=no_mcp,
trust_project_mcp=trust_project_mcp,
@@ -413,6 +419,8 @@ async def server_session(
enable_interpreter: bool | None = None,
interpreter_ptc: str | list[str] | None = None,
interpreter_ptc_acknowledge_unsafe: bool = False,
rubric_model: str | None = None,
rubric_max_iterations: int = 3,
mcp_config_path: str | None = None,
no_mcp: bool = False,
trust_project_mcp: bool | None = None,
@@ -443,6 +451,8 @@ async def server_session(
interpreter_ptc: Override for `settings.interpreter_ptc` (PTC allowlist).
interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for
`interpreter_ptc="all"` outside of `auto_approve`.
rubric_model: Grader model spec; `None` reuses the main model.
rubric_max_iterations: Grader iterations per rubric attempt.
mcp_config_path: Path to MCP config.
no_mcp: Disable MCP.
trust_project_mcp: Trust project MCP servers.
@@ -474,6 +484,8 @@ async def server_session(
enable_interpreter=enable_interpreter,
interpreter_ptc=interpreter_ptc,
interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe,
rubric_model=rubric_model,
rubric_max_iterations=rubric_max_iterations,
mcp_config_path=mcp_config_path,
no_mcp=no_mcp,
trust_project_mcp=trust_project_mcp,
libs/code/deepagents_code/ui.py
+11
View File
@@ -185,6 +185,17 @@ def show_help() -> None:
console.print(
" --max-turns N Max agentic turns before stopping (needs -n)"
)
console.print(
" --rubric TEXT|@PATH Acceptance criteria to grade against; "
"'@path' reads a file relative to cwd, '~' ok (needs -n)"
)
console.print(
" --rubric-model MODEL Model the rubric grader uses "
"(defaults to main model)"
)
console.print(
" --rubric-max-iterations N Grader iterations per rubric attempt (default 3)"
)
console.print(
" --timeout SECONDS Hard wall-clock limit; exits 124 on expiry"
" (needs -n/stdin)"
libs/code/tests/unit_tests/test_agent.py
+38
View File
@@ -3180,6 +3180,44 @@ class TestCreateCliAgentInterpreterWiring:
mock_settings.interpreter_ptc_acknowledge_unsafe = False
return mock_settings
def test_appends_rubric_middleware(self, tmp_path: Path) -> None:
from deepagents.middleware.rubric import RubricMiddleware
mock_settings = self._build_mock_settings(tmp_path)
mock_agent = Mock()
mock_agent.with_config.return_value = mock_agent
fake_model = _make_fake_chat_model()
with (
patch("deepagents_code.agent.settings", mock_settings),
patch("deepagents_code.agent.SkillsMiddleware"),
patch("deepagents_code.agent.MemoryMiddleware"),
patch(
"deepagents_code.agent.create_deep_agent",
return_value=mock_agent,
) as mock_create,
patch(
"deepagents._models.init_chat_model",
return_value=fake_model,
),
):
create_cli_agent(
model="fake-model",
assistant_id="test",
enable_memory=False,
enable_skills=False,
enable_shell=False,
rubric_model="custom-grader-model",
rubric_max_iterations=5,
)
_, kwargs = mock_create.call_args
rubrics = [
mw for mw in kwargs["middleware"] if isinstance(mw, RubricMiddleware)
]
assert len(rubrics) == 1
assert rubrics[0]._model == "custom-grader-model"
assert rubrics[0].max_iterations == 5
def test_appends_interpreter_middleware_when_enabled(self, tmp_path: Path) -> None:
from langchain_quickjs import CodeInterpreterMiddleware
libs/code/tests/unit_tests/test_rubric_cli.py
+237
View File
@@ -0,0 +1,237 @@
"""Unit tests for rubric (`RubricMiddleware`) CLI wiring."""
from __future__ import annotations
import io
import os
import subprocess
import sys
import textwrap
from typing import TYPE_CHECKING
from unittest.mock import patch
import pytest
from rich.console import Console
if TYPE_CHECKING:
from pathlib import Path
from deepagents_code._env_vars import SERVER_ENV_PREFIX
from deepagents_code._server_config import ServerConfig
from deepagents_code.main import _resolve_rubric_text
from deepagents_code.non_interactive import (
StreamState,
_build_non_interactive_header,
_process_rubric_event,
)
class TestResolveRubricText:
"""`_resolve_rubric_text` literal/file/@path resolution."""
def test_none_when_unset(self) -> None:
assert _resolve_rubric_text(None) is None
def test_literal(self) -> None:
assert _resolve_rubric_text("tests pass; minimal") == "tests pass; minimal"
def test_literal_is_stripped(self) -> None:
assert _resolve_rubric_text(" do X ") == "do X"
def test_empty_literal_rejected(self) -> None:
with pytest.raises(ValueError, match="must not be empty"):
_resolve_rubric_text(" ")
def test_at_path_in_rubric(self, tmp_path: Path) -> None:
f = tmp_path / "rubric.md"
f.write_text("from at-path", encoding="utf-8")
assert _resolve_rubric_text(f"@{f}") == "from at-path"
def test_at_prefix_always_treated_as_path(self) -> None:
# Documents the one-way ambiguity: any `@`-prefixed value is read as a
# file path, so a literal rubric beginning with `@` is unreachable and
# surfaces a read error rather than being used verbatim.
with pytest.raises(ValueError, match="Could not read rubric file"):
_resolve_rubric_text("@tests pass; minimal diff")
def test_bare_at_sign_rejected(self) -> None:
# `@` with no path (e.g. an empty shell glob) must still error rather
# than silently resolving to the current directory.
with pytest.raises(ValueError, match="Could not read rubric file"):
_resolve_rubric_text("@")
def test_at_path_expands_tilde(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
# Exercises the `.expanduser()` call, otherwise uncovered.
monkeypatch.setenv("HOME", str(tmp_path))
(tmp_path / "rubric.md").write_text("tilde criteria", encoding="utf-8")
assert _resolve_rubric_text("@~/rubric.md") == "tilde criteria"
def test_missing_file(self, tmp_path: Path) -> None:
with pytest.raises(ValueError, match="Could not read rubric file"):
_resolve_rubric_text(f"@{tmp_path / 'nope.md'}")
def test_empty_file(self, tmp_path: Path) -> None:
f = tmp_path / "rubric.md"
f.write_text(" \n", encoding="utf-8")
with pytest.raises(ValueError, match="is empty"):
_resolve_rubric_text(f"@{f}")
def _run_cli_main_devnull_stdin(argv: list[str]) -> subprocess.CompletedProcess[str]:
"""Run `cli_main` in a subprocess with empty (non-piped) stdin.
`stdin=DEVNULL` makes `apply_stdin_pipe` read an empty string and return
early, so `non_interactive_message` stays unset — the deterministic way to
reach the interactive-only argument guards without a TTY. `parse_args`
handles `--non-interactive`/`-m`, and `check_cli_dependencies` is patched
purely for environment portability (it only calls `importlib.util.find_spec`).
"""
code = """
import sys
from unittest.mock import patch
from deepagents_code.main import cli_main
with (
patch.object(sys, "argv", sys.argv[1:]),
patch("deepagents_code.main.check_cli_dependencies"),
):
cli_main()
"""
return subprocess.run(
[sys.executable, "-c", textwrap.dedent(code), "deepagents", *argv],
capture_output=True,
text=True,
stdin=subprocess.DEVNULL,
timeout=30,
check=False,
)
class TestRubricGating:
"""Rubric flags require `-n`/piped stdin; the guard lives in `cli_main`."""
def test_rubric_without_non_interactive_errors(self) -> None:
result = _run_cli_main_devnull_stdin(["--rubric", "tests pass"])
assert result.returncode == 2, result.stderr
assert "--non-interactive" in result.stderr
assert "--rubric" in result.stderr
# The removed flag must not resurface in the guidance.
assert "--rubric-file" not in result.stderr
class TestServerConfigRubric:
"""Rubric grader settings round-trip through env serialization."""
def test_defaults(self) -> None:
config = ServerConfig()
assert config.rubric_model is None
assert config.rubric_max_iterations == 3
def test_round_trip(self) -> None:
original = ServerConfig(
rubric_model="anthropic:claude-sonnet-4-6",
rubric_max_iterations=5,
)
env = {
f"{SERVER_ENV_PREFIX}{k}": v
for k, v in original.to_env().items()
if v is not None
}
with patch.dict(os.environ, env, clear=False):
restored = ServerConfig.from_env()
assert restored.rubric_model == "anthropic:claude-sonnet-4-6"
assert restored.rubric_max_iterations == 5
def test_from_cli_args_forwards_rubric_settings(self) -> None:
config = ServerConfig.from_cli_args(
project_context=None,
model_name=None,
model_params=None,
assistant_id="agent",
auto_approve=False,
sandbox_type="none",
sandbox_id=None,
sandbox_snapshot_name=None,
sandbox_setup=None,
enable_shell=True,
enable_ask_user=False,
rubric_model="openai:gpt-5.1",
rubric_max_iterations=7,
mcp_config_path=None,
no_mcp=False,
trust_project_mcp=None,
interactive=True,
)
assert config.rubric_model == "openai:gpt-5.1"
assert config.rubric_max_iterations == 7
class TestHeaderIndicator:
def test_rubric_active_marker(self) -> None:
header = _build_non_interactive_header("agent", "thread-1", rubric_active=True)
assert "Rubric: active" in header.plain
def test_no_marker_when_inactive(self) -> None:
header = _build_non_interactive_header("agent", "thread-1", rubric_active=False)
assert "Rubric" not in header.plain
def _render_event(data: dict) -> str:
state = StreamState()
buf = io.StringIO()
console = Console(file=buf, width=200, highlight=False)
_process_rubric_event(data, state, console)
return buf.getvalue()
class TestProcessRubricEvent:
def test_ignores_non_rubric_payload(self) -> None:
assert _render_event({"type": "something_else"}) == ""
def test_start_event(self) -> None:
out = _render_event({"type": "rubric_evaluation_start", "iteration": 0})
assert "Grading against rubric" in out
assert "iteration 1" in out
def test_satisfied(self) -> None:
out = _render_event(
{"type": "rubric_evaluation_end", "result": "satisfied", "criteria": []}
)
assert "Rubric satisfied" in out
def test_needs_revision_with_criteria(self) -> None:
out = _render_event(
{
"type": "rubric_evaluation_end",
"result": "needs_revision",
"explanation": "tests missing",
"criteria": [
{"name": "tests", "passed": False, "gap": "no coverage"},
{"name": "style", "passed": True},
],
}
)
assert "needs revision" in out
assert "tests missing" in out
assert "no coverage" in out
assert "style" not in out
def test_max_iterations(self) -> None:
out = _render_event(
{"type": "rubric_evaluation_end", "result": "max_iterations_reached"}
)
assert "max iterations reached" in out
def test_failed(self) -> None:
out = _render_event(
{
"type": "rubric_evaluation_end",
"result": "failed",
"explanation": "bad rubric",
}
)
assert "grader failed" in out
assert "bad rubric" in out
libs/code/tests/unit_tests/test_server_config.py
+42
View File
@@ -14,6 +14,7 @@ from deepagents_code._server_config import (
_interpreter_suppressed_by_sandbox,
_normalize_path,
_read_env_bool,
_read_env_int,
_read_env_json,
_read_env_optional_bool,
_read_env_str,
@@ -88,6 +89,25 @@ class TestReadEnvJson:
_read_env_json("DATA")
# ------------------------------------------------------------------
# _read_env_int
# ------------------------------------------------------------------
class TestReadEnvInt:
def test_valid_int(self) -> None:
with patch.dict(os.environ, {f"{SERVER_ENV_PREFIX}COUNT": "5"}):
assert _read_env_int("COUNT", default=3) == 5
def test_missing_returns_default(self) -> None:
with patch.dict(os.environ, {}, clear=True):
assert _read_env_int("COUNT", default=3) == 3
def test_malformed_returns_default(self) -> None:
with patch.dict(os.environ, {f"{SERVER_ENV_PREFIX}COUNT": "abc"}):
assert _read_env_int("COUNT", default=3) == 3
# ------------------------------------------------------------------
# _read_env_str
# ------------------------------------------------------------------
@@ -355,6 +375,28 @@ class TestServerConfigEdgeCases:
assert restored.enable_interpreter is False
def test_malformed_rubric_max_iterations_env_uses_default(self) -> None:
"""Bad optional rubric iteration config must not break graph startup."""
with patch.dict(
os.environ,
{f"{SERVER_ENV_PREFIX}RUBRIC_MAX_ITERATIONS": "abc"},
clear=True,
):
restored = ServerConfig.from_env()
assert restored.rubric_max_iterations == 3
def test_rubric_max_iterations_env_parses_int(self) -> None:
"""Valid rubric iteration config survives the server env boundary."""
with patch.dict(
os.environ,
{f"{SERVER_ENV_PREFIX}RUBRIC_MAX_ITERATIONS": "7"},
clear=True,
):
restored = ServerConfig.from_env()
assert restored.rubric_max_iterations == 7
def test_sandbox_type_none_string_round_trips(self) -> None:
"""sandbox_type='none' normalizes to None and survives round-trip."""
original = ServerConfig(sandbox_type="none")
libs/code/tests/unit_tests/test_server_graph.py
+2
View File
@@ -202,6 +202,8 @@ class TestServerGraph:
enable_skills=True,
enable_shell=True,
enable_interpreter=False,
rubric_model=None,
rubric_max_iterations=3,
mcp_server_info=mcp_server_info,
cwd=None,
project_context=None,