fix(tools): fit structured evidence to consumer limits and file boundaries

Signed-off-by: Haozhe Wu <haozhe_wu@connect.hku.hk>
This commit is contained in:
Haozhe Wu committed 2026-10-04 03:09:29 +08:00
1 parent 3ee353f719
commit 8ec3b80f10
11 files changed
+378 -141

No files matched your search

+18 -10
View File
@@ -99,12 +99,21 @@ def _merge_cjk_group(raw_tokens: list[str], inners: list[str]) -> str:
return "".join(unigrams)
chars: list[str] = []
for inner in inners:
highlighted: list[bool] = []
for raw, inner in zip(raw_tokens, inners):
marked = ">>>" in raw or "<<<" in raw
if chars and chars[-1] == inner[0]:
highlighted[-1] = highlighted[-1] or marked
chars.append(inner[1])
highlighted.append(marked)
else:
chars.extend(inner)
return "".join(chars)
highlighted.extend([marked] * len(inner))
rendered = "".join(f">>>{char}<<<" if marked else char
for char, marked in zip(chars, highlighted))
return ("..." if raw_tokens[0].startswith("...") else "") + rendered + (
"..." if raw_tokens[-1].endswith("...") else ""
)
class _CleanPiece:
@@ -499,7 +508,6 @@ class MemorySearchIndex:
next_stripped = _strip_markers(nxt.text)
ends_cjk = bool(stripped) and _is_cjk_char(stripped[-1])
next_starts_cjk = bool(next_stripped) and _is_cjk_char(next_stripped[0])
has_markers = piece.text != stripped or nxt.text != next_stripped
if ends_cjk and next_starts_cjk:
# Both sides read as CJK script once markers are stripped
@@ -509,7 +517,7 @@ class MemorySearchIndex:
# _prepare_cjk) earns a space back.
if piece.gap_after >= 2:
joined.append(" ")
elif not has_markers and ends_cjk != next_starts_cjk:
elif ends_cjk != next_starts_cjk:
# A plain CJK run directly against plain non-CJK text (a
# ticker, a date, punctuation) -- same artificial-join logic,
# just across the script boundary instead of within it.
@@ -527,20 +535,20 @@ class MemorySearchIndex:
@staticmethod
def _sanitize_fts_query(query: str) -> str:
"""Sanitize user query for FTS5 MATCH syntax.
Extracts alphanumeric tokens (2+ chars) and CJK characters,
generates bigrams for consecutive CJK chars, quotes each token
and joins with OR to prevent FTS5 operator injection.
Args:
query: Raw user query string.
Returns:
FTS5-safe MATCH expression, or empty-quoted string if no tokens.
"""
tokens: list[str] = []
cjk_buffer: list[str] = []
# Walk through pre-extracted raw tokens
raw_tokens = re.findall(
r"[a-zA-Z0-9_]{2,}|[\u4e00-\u9fff\u3400-\u4dbf]", query
@@ -553,10 +561,10 @@ class MemorySearchIndex:
tokens.extend(_cjk_query_tokens(cjk_buffer))
cjk_buffer = []
tokens.append(tok)
if cjk_buffer:
tokens.extend(_cjk_query_tokens(cjk_buffer))
if not tokens:
return '""'
# Quote each token and join with OR for broader matching
+28 -23
View File
@@ -27,7 +27,7 @@ from typing import Any, Mapping
logger = logging.getLogger(__name__)
#: Version of the summary contract consumed by the frontend.
#: Version of the structured tool-result summary contract.
SCHEMA_VERSION = "1.0"
#: Hard cap on ``equity_preview`` points; longer curves are equal-stride
#: sampled with the first and last row always included.
@@ -61,7 +61,7 @@ def collect_ohlcv_paths(run_dir: Path) -> dict[str, str]:
candidates = sorted((Path(run_dir) / "artifacts").glob(_OHLCV_GLOB))
for path in candidates:
code = path.name[len(_OHLCV_PREFIX) : -len(_OHLCV_SUFFIX)]
if code:
if code and path.is_file() and path.resolve().is_relative_to(Path(run_dir).resolve()):
mapping[code] = str(path)
except OSError as exc: # pragma: no cover - defensive, glob rarely raises
logger.debug("ohlcv glob failed for %s: %s", run_dir, exc)
@@ -94,7 +94,7 @@ def build_backtest_summary(
json.JSONDecodeError: ``run_card.json`` is not valid JSON.
ValueError: ``run_card.json`` does not contain a JSON object.
"""
run_dir = Path(run_dir)
run_dir = Path(run_dir).resolve()
card = _load_run_card(run_dir)
backtest = card.get("backtest")
if not isinstance(backtest, dict):
@@ -106,7 +106,10 @@ def build_backtest_summary(
ohlcv_paths = collect_ohlcv_paths(run_dir)
artifacts_dir = run_dir / "artifacts"
return {
for name in ("equity.csv", "trades.csv", "metrics.csv"):
_contained(run_dir, artifacts_dir / name)
ohlcv_paths = {code: str(_contained(run_dir, Path(path))) for code, path in ohlcv_paths.items()}
return _json_safe({
"schema_version": SCHEMA_VERSION,
"run_id": run_dir.name,
"title": _TITLE_SEPARATOR.join(codes),
@@ -116,6 +119,8 @@ def build_backtest_summary(
"interval": backtest.get("interval"),
"initial_cash": backtest.get("initial_cash"),
"metrics": _json_safe(_as_dict(card.get("metrics"))),
"structured_metrics": _json_safe(_as_dict(card.get("structured_metrics"))),
"validation": _json_safe(_as_dict(card.get("validation"))),
"equity_preview": _equity_preview(artifacts_dir / "equity.csv"),
"artifact_paths": {
"equity": _existing_file(artifacts_dir / "equity.csv"),
@@ -127,7 +132,7 @@ def build_backtest_summary(
"warnings": _json_safe(
card.get("warnings") if isinstance(card.get("warnings"), list) else []
),
}
})
def try_build_backtest_summary(
@@ -154,7 +159,7 @@ def try_build_backtest_summary(
def _load_run_card(run_dir: Path) -> dict[str, Any]:
"""Read and validate ``run_card.json`` at the run directory root."""
payload = json.loads((run_dir / "run_card.json").read_text(encoding="utf-8"))
payload = json.loads(_contained(run_dir, run_dir / "run_card.json").read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError(
f"run_card.json must contain a JSON object, got {type(payload).__name__}"
@@ -162,6 +167,14 @@ def _load_run_card(run_dir: Path) -> dict[str, Any]:
return payload
def _contained(root: Path, path: Path) -> Path:
"""Refuse artifacts whose resolved destination escapes the run directory."""
resolved = path.resolve()
if not resolved.is_relative_to(root.resolve()):
raise ValueError("Summary artifact resolves outside run directory")
return path
def _as_dict(value: Any) -> dict[str, Any]:
"""Return ``value`` when it is a dict, else an empty dict."""
return value if isinstance(value, dict) else {}
@@ -218,26 +231,18 @@ def _equity_preview(equity_path: Path) -> list[dict[str, Any]]:
"""
if not equity_path.is_file():
return []
with equity_path.open(encoding="utf-8", newline="") as stream:
total = sum(1 for _ in csv.DictReader(stream))
indices = set(range(total)) if total <= MAX_PREVIEW_POINTS else {
i * (total - 1) // (MAX_PREVIEW_POINTS - 1) for i in range(MAX_PREVIEW_POINTS)
}
rows: list[dict[str, Any]] = []
with equity_path.open(encoding="utf-8", newline="") as stream:
for record in csv.DictReader(stream):
for index, record in enumerate(csv.DictReader(stream)):
if index not in indices:
continue
point: dict[str, Any] = {"time": record.get("timestamp")}
for field in _PREVIEW_NUMERIC_FIELDS:
point[field] = _parse_float(record.get(field))
rows.append(point)
return _sample_rows(rows, MAX_PREVIEW_POINTS)
def _sample_rows(rows: list[dict[str, Any]], limit: int) -> list[dict[str, Any]]:
"""Equal-stride sample of at most ``limit`` rows, endpoints always kept.
Rows within the limit are returned unchanged. Beyond it, the integer
stride ``i * (n - 1) // (limit - 1)`` yields exactly ``limit`` distinct
indices that include the first (``i = 0``) and last (``i = limit - 1``)
row by construction.
"""
total = len(rows)
if limit < 2 or total <= limit:
return list(rows)
indices = (i * (total - 1) // (limit - 1) for i in range(limit))
return [rows[index] for index in indices]
return rows
+32 -4
View File
@@ -11,6 +11,7 @@ from backtest.loaders.registry import VALID_SOURCES
from src.agent.progress import emit_progress
from src.agent.tools import BaseTool
from src.config.accessor import get_env_config
from src.config.limits import TOOL_RESULT_LIMIT
from src.core.runner import Runner
from src.core.state import RunStateStore
from src.tools.backtest_summary import collect_ohlcv_paths, try_build_backtest_summary
@@ -163,15 +164,42 @@ def run_backtest(run_dir: str) -> str:
"run_dir": run_dir,
}
if result.success:
# Strictly additive best-effort fields: the legacy keys above stay
# byte-identical, and a summary that cannot be built (missing or
# corrupt run_card.json) is omitted rather than failing the run.
# Small envelopes retain the legacy fields unchanged. Oversized
# envelopes are fitted below with explicit omission notices. A missing
# or corrupt run card never turns a successful run into an error.
ohlcv_paths = collect_ohlcv_paths(run_path)
artifacts_found["ohlcv"] = ohlcv_paths
summary = try_build_backtest_summary(run_path, ohlcv_paths)
if summary is not None:
envelope["summary"] = summary
return json.dumps(envelope, ensure_ascii=False)
return _serialize_result(envelope)
def _serialize_result(envelope: dict[str, Any]) -> str:
"""Keep structured results usable through the agent's delivery cap."""
def encode() -> str:
return json.dumps(envelope, ensure_ascii=False)
if len(encode()) <= TOOL_RESULT_LIMIT:
return encode()
for key in ("stdout", "stderr"):
if envelope.get(key):
envelope[key] = ""
envelope["logs_omitted"] = "Read the run logs for full stdout/stderr."
summary = envelope.get("summary")
if isinstance(summary, dict):
points = summary.get("equity_preview", [])
while len(encode()) > TOOL_RESULT_LIMIT and len(points) > 2:
count = max(2, len(points) // 2)
points = [points[i * (len(points) - 1) // (count - 1)] for i in range(count)]
summary["equity_preview"] = points
if len(encode()) > TOOL_RESULT_LIMIT:
envelope.pop("summary")
envelope["summary_omitted"] = "Summary exceeds the result budget; inspect run_card.json locally or use read_run_artifact in meta mode."
if len(encode()) > TOOL_RESULT_LIMIT:
envelope["artifacts"] = {}
envelope["artifacts_omitted"] = "Artifact paths exceed the result budget; inspect the run directory with read_file."
return encode()
class BacktestTool(BaseTool):
+35 -44
View File
@@ -25,6 +25,7 @@ from typing import Any
from backtest.loaders._http import resolve_min_interval, throttled_get_json
from src.agent.tools import BaseTool
from src.config.accessor import get_env_config
from src.config.limits import TOOL_RESULT_LIMIT
logger = logging.getLogger(__name__)
@@ -91,7 +92,8 @@ class FredMacroTool(BaseTool):
"series is capped rather than refused: the envelope "
"reports truncated and observations_available, plus a hint "
"naming the way out (raise this where the cap allows it, "
"or narrow the date window)."
"or narrow the date window). The result character budget "
"can return fewer observations than this limit."
),
"default": _DEFAULT_LIMIT,
},
@@ -125,9 +127,9 @@ class FredMacroTool(BaseTool):
The series is capped at ``limit`` observations, keeping the most
recent ones; ``truncated`` says whether that cap bit,
``observations_available`` is the count before capping, and a
``hint`` names the way out. The scalars are emitted before
``observations`` so a capped series still reports itself after the
tool-result character cap trims the array. On failure:
``hint`` names the way out. Whole observations also fit the shared
character budget, so count is the number actually delivered.
The hint distinguishes that budget from the row limit. On failure:
``{"ok": false, "error": str}``.
"""
api_key = get_env_config().data.fred_api_key or None
@@ -169,48 +171,37 @@ class FredMacroTool(BaseTool):
return _error(f"no observations found for series '{series_id}'")
limit = _clamp_limit(kwargs.get("limit", _DEFAULT_LIMIT))
# Keep the most recent observations when the cap is exceeded; FRED serves
# oldest-first, so the tail holds the newest records.
capped = observations[-limit:]
# A capped series must not read as a complete one: the caller asked for a
# date window and would otherwise take the returned tail for the whole
# window. Mirrors the convention get_market_data's own cap uses (cap_rows
# in agent/src/market_data.py): flag it, and say the way out.
available = len(observations)
truncated = available > limit
data: dict[str, Any] = {
"series_id": series_id,
"count": len(capped),
"observations_available": available,
"truncated": truncated,
"limit": limit,
}
if truncated:
# At the ceiling, "raise limit" is advice the caller cannot take.
if limit < _MAX_LIMIT:
remedy = f"raise limit (max {_MAX_LIMIT}) or narrow the date window"
else:
upstream_count = payload.get("count") if isinstance(payload, dict) else None
if isinstance(upstream_count, int) and upstream_count > len(payload.get("observations", [])):
return _error("FRED returned only part of the requested date window; narrow the date window and retry")
count = min(limit, available)
while True:
capped = observations[-count:]
truncated = count < available
data: dict[str, Any] = {
"series_id": series_id,
"count": count,
"observations_available": available,
"truncated": truncated,
"limit": limit,
}
if truncated:
reason = "result character budget" if count < min(limit, available) else "observation limit"
remedy = "narrow the date window"
data["hint"] = (
f"returned the {limit} most recent of {available} observations; "
f"{remedy}"
)
# The array goes last so the report above survives the tool-result
# character cap (TOOL_RESULT_LIMIT, src/config/limits.py), which cuts an
# oversized envelope to a prefix. Behind 2,000 rows of observations the
# cap report never reaches the caller it exists for.
data["observations"] = capped
return json.dumps(
{
"ok": True,
"market": "US",
"source": "fred",
"data": data,
},
ensure_ascii=False,
)
if reason == "observation limit" and limit < _MAX_LIMIT:
remedy = f"raise limit (max {_MAX_LIMIT}) or {remedy}"
data["hint"] = (
f"returned the {count} most recent of {available} observations "
f"due to the {reason}; {remedy}"
)
data["observations"] = capped
result = json.dumps({"ok": True, "market": "US", "source": "fred", "data": data}, ensure_ascii=False)
if len(result) <= TOOL_RESULT_LIMIT:
return result
if count <= 1:
return _error("One observation exceeds the result budget; request another series or date window")
count = max(1, min(count - 1, int(count * TOOL_RESULT_LIMIT / len(result))))
def _parse_observations(payload: Any) -> list[dict]:
+86 -47
View File
@@ -4,7 +4,7 @@ A backtest run writes CSV artifacts (``equity.csv``, ``trades.csv``, ...) plus
a ``run_card.json`` sidecar whose ``artifacts`` manifest lists every file the
run produced with its sha256 and size. The generic ``read_file`` tool returns
raw text, burning LLM tokens on comma-separated noise and truncating mid-file.
This tool parses the artifact once and serves structured JSON pages:
This tool scans artifacts without retaining the full table and serves structured JSON pages:
* ``rows`` — offset paging over whole records with an honest ``truncated`` /
``next_offset`` contract, so a walk reassembles the file losslessly.
@@ -33,12 +33,11 @@ from pathlib import Path
from typing import Any
from src.agent.tools import BaseTool
from src.config.limits import TOOL_RESULT_LIMIT
from src.tools.path_utils import safe_run_dir
# Serialized-envelope budget in characters. MCP clients truncate large tool
# results (50KB is a common default; raised harness limits reach 256KB); 120K
# keeps a full page intact under the raised limit while leaving framing room.
_BYTE_BUDGET = 120_000
# Match the cap applied by both the agent and swarm consumers.
_BYTE_BUDGET = TOOL_RESULT_LIMIT
# Friendly CSV aliases: name -> path relative to run_dir.
_CSV_ARTIFACTS: dict[str, str] = {
@@ -133,7 +132,10 @@ def _load_artifact_manifest(run_root: Path) -> dict[str, tuple[Path, dict[str, A
when the card carries no well-formed ``artifacts`` list.
"""
try:
card = json.loads((run_root / _RUN_CARD_PATH).read_text(encoding="utf-8"))
card_path = _safe_relative_resolve(run_root, _RUN_CARD_PATH)
if card_path is None:
return {}
card = json.loads(card_path.read_text(encoding="utf-8"))
except (OSError, ValueError, UnicodeDecodeError):
return {}
entries = card.get("artifacts") if isinstance(card, dict) else None
@@ -220,6 +222,8 @@ def _coerce_value(raw: str) -> Any:
text = raw.strip()
if not text:
return None
if re.fullmatch(r"[+-]?0[0-9]+", text):
return raw
if "_" not in text:
try:
return int(text)
@@ -233,26 +237,29 @@ def _coerce_value(raw: str) -> Any:
return raw
def _read_csv(path: Path) -> tuple[list[str], list[list[Any]]]:
"""Parse a CSV artifact into columns plus coerced, header-aligned rows.
def _csv_shape(path: Path) -> tuple[list[str], int]:
"""Count records without retaining the source table in memory."""
with path.open(encoding="utf-8-sig", newline="") as stream:
reader = csv.reader(stream)
header = next(reader, [])
return header, sum(1 for row in reader if row)
Each row has exactly ``len(columns)`` cells (short rows padded with
``None``, extra trailing cells dropped); an empty file yields ``([], [])``.
"""
with path.open("r", encoding="utf-8-sig", newline="") as handle:
reader = csv.reader(handle)
try:
header = next(reader)
except StopIteration:
return [], []
width = len(header)
rows: list[list[Any]] = []
def _read_csv(path: Path, indices: set[int], width: int) -> list[list[Any]]:
"""Read only selected records, retaining at most the requested page."""
rows: list[list[Any]] = []
with path.open(encoding="utf-8-sig", newline="") as stream:
reader = csv.reader(stream)
next(reader, None)
index = 0
for record in reader:
if not record:
continue
aligned = record[:width] + [""] * (width - len(record))
rows.append([_coerce_value(cell) for cell in aligned])
return header, rows
if index in indices:
aligned = record[:width] + [""] * max(0, width - len(record))
rows.append([_coerce_value(cell) for cell in aligned])
index += 1
return rows
def _project(
@@ -309,8 +316,8 @@ def _fit_rows_payload(
overflow (then one further) so wide records still converge, keeping
``returned_rows`` / ``truncated`` / ``next_offset`` honest — a shrunk page
always reports ``truncated: true`` with the resume offset, so the tail is
never dropped silently. The payload stays valid JSON even when a single
row alone exceeds the budget.
never dropped silently. A record too large to fit returns an actionable
error instead of an oversized payload.
"""
count = len(page)
while True:
@@ -325,25 +332,31 @@ def _fit_rows_payload(
"next_offset": offset + returned if truncated else None,
}
payload = _serialize(envelope)
if len(payload) <= budget or count <= 1:
if len(payload) <= budget:
return payload
if count <= 1:
return _error("A single record exceeds the result budget",
"Project fewer columns, or use read_file for this record.")
count = max(1, min(count - 1, int(count * budget / len(payload))))
def _fit_downsample_payload(
base: dict[str, Any], rows: list[list[Any]], max_points: int, budget: int
base: dict[str, Any], path: Path, header: list[str],
requested: list[str] | None, max_points: int, budget: int
) -> str:
"""Serialize a downsample envelope, re-striding to fewer points if needed.
``base`` is the envelope skeleton without ``rows`` / ``downsample``
fields; ``rows`` are the parsed (and projected) source rows. The first
fields. Each pass reads only selected rows from ``path``; the first
and last source row stay pinned at every sample size.
"""
total = len(rows)
target = max_points
total = base["total_rows"]
target = max(2, max_points)
while True:
indices, stride = _downsample_indices(total, target)
sample = [rows[i] for i in indices]
sample = _read_csv(path, set(indices), len(header))
if requested is not None:
_, sample = _project(header, sample, requested)
envelope = {
**base,
"rows": sample,
@@ -358,25 +371,38 @@ def _fit_downsample_payload(
},
}
payload = _serialize(envelope)
if len(payload) <= budget or target <= 2:
if len(payload) <= budget:
return payload
if target <= 2:
return _error("The sample endpoints exceed the result budget",
"Project fewer columns, or request individual rows.")
target = max(2, min(target - 1, int(target * budget / len(payload))))
def _read_json_artifact(path: Path, artifact: str, run_dir: str) -> str:
def _read_json_artifact(path: Path, artifact: str, run_dir: str, *, meta: bool = False) -> str:
"""Parse and envelope a small JSON artifact (run_card or manifest-listed).
Returns the ``{"artifact", "run_dir", "json", "size_bytes"}`` envelope, or
the error envelope when the file is corrupt — never partial JSON.
"""
try:
parsed = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError) as exc:
with path.open(encoding="utf-8") as stream:
content = stream.read(_BYTE_BUDGET + 1)
if len(content) > _BYTE_BUDGET:
if meta:
return _serialize({"artifact": artifact, "run_dir": run_dir,
"size_bytes": path.stat().st_size, "format": "json",
"content_omitted": True,
"hint": "JSON exceeds the result budget; inspect it locally."})
return _error("JSON artifact exceeds the result budget",
"Inspect the artifact locally or regenerate a smaller JSON artifact.")
parsed = json.loads(content, parse_constant=lambda value: None)
except (OSError, json.JSONDecodeError, UnicodeDecodeError) as exc:
return _error(
f"{artifact} is not valid JSON: {exc}",
f"Delete or regenerate {path.name}; this tool never returns partial JSON.",
)
return _serialize(
payload = _serialize(
{
"artifact": artifact,
"run_dir": run_dir,
@@ -385,6 +411,10 @@ def _read_json_artifact(path: Path, artifact: str, run_dir: str) -> str:
}
)
if len(payload) > _BYTE_BUDGET:
return _error("JSON artifact exceeds the result budget", "Inspect the artifact locally or regenerate a smaller JSON artifact.")
return payload
def _meta_envelope(
base: dict[str, Any],
@@ -411,7 +441,10 @@ def _meta_envelope(
declared = entry.get("size_bytes")
if isinstance(declared, int) and not isinstance(declared, bool):
envelope["manifest_size_bytes"] = declared
return _serialize(envelope)
payload = _serialize(envelope)
if len(payload) > _BYTE_BUDGET:
return _error("Artifact metadata exceeds the result budget", "Project fewer columns.")
return payload
def read_run_artifact(
@@ -433,9 +466,10 @@ def read_run_artifact(
artifacts manifest (e.g. ``artifacts/validation.json``).
format: ``rows`` (offset paging), ``downsample`` (equal-stride sample,
first+last pinned; ``offset`` ignored) or ``meta`` (shape only).
JSON artifacts return their parsed object regardless of format.
JSON artifacts return their parsed object when it fits; oversized
JSON returns size metadata only in meta mode, otherwise an error.
offset: First row index for ``rows`` mode; negative values refused.
max_rows: Page/sample size, clamped to ``[1, 5000]``.
max_rows: Page size clamped to ``[1, 5000]``; samples need at least two endpoints.
columns: Optional projection applied before budget fitting; an unknown
name is refused with the valid column list.
@@ -456,7 +490,7 @@ def read_run_artifact(
except ValueError as exc:
return _error(str(exc), "Pass the run_dir a backtest/tool call returned.")
manifest = _load_artifact_manifest(run_root)
manifest = {} if artifact == _RUN_CARD_ALIAS else _load_artifact_manifest(run_root)
try:
path, entry = _resolve_artifact(run_root, artifact, manifest)
except _ArtifactError as exc:
@@ -470,7 +504,7 @@ def read_run_artifact(
)
if path.suffix.lower() == ".json":
return _read_json_artifact(path, artifact, run_dir)
return _read_json_artifact(path, artifact, run_dir, meta=format == "meta")
try:
offset = int(offset)
@@ -488,7 +522,7 @@ def read_run_artifact(
max_rows = max(1, min(max_rows, _MAX_ROWS_CEILING))
try:
header, rows = _read_csv(path)
header, total_rows = _csv_shape(path)
except (OSError, UnicodeDecodeError, csv.Error) as exc:
return _error(
f"failed to read {artifact}: {exc}",
@@ -496,15 +530,17 @@ def read_run_artifact(
)
if columns is not None:
try:
header, rows = _project(header, rows, list(columns))
_project(header, [], list(columns))
except ValueError as exc:
return _error(
str(exc),
"Pass only column names from the artifact header, or omit 'columns'.",
)
total_rows = len(rows)
base: dict[str, Any] = {"artifact": artifact, "run_dir": run_dir, "columns": header}
source_header = header
if columns is not None:
header = list(columns)
base: dict[str, Any] = {"artifact": artifact, "run_dir": run_dir, "path": str(path), "columns": header}
budget = _BYTE_BUDGET
if format == "meta":
@@ -512,11 +548,14 @@ def read_run_artifact(
if format == "downsample":
return _fit_downsample_payload(
{**base, "total_rows": total_rows, "offset": 0}, rows, max_rows, budget
{**base, "total_rows": total_rows, "offset": 0},
path, source_header, columns, max_rows, budget
)
# rows mode
page = rows[offset : offset + max_rows] if offset < total_rows else []
page = _read_csv(path, set(range(offset, min(total_rows, offset + max_rows))), len(source_header))
if columns is not None:
_, page = _project(source_header, page, columns)
envelope_base = {
**base,
"total_rows": total_rows,
@@ -570,7 +609,7 @@ class RunArtifactTool(BaseTool):
},
"max_rows": {
"type": "integer",
"description": "Page/sample size, clamped to [1, 5000] (default 1000)",
"description": "Page size [1, 5000]; samples keep at least two endpoints (default 1000)",
},
"columns": {
"type": "array",
+10 -4
View File
@@ -321,7 +321,7 @@ class SkillFileTool(BaseTool):
# Validate subdirectory
parts = Path(rel_path).parts
if len(parts) < 2 or parts[0] not in _ALLOWED_SUBDIRS:
if len(parts) < 2 or parts[0] not in _ALLOWED_SUBDIRS or ".." in parts:
return json.dumps({
"status": "error",
"error": f"Path must start with one of: {', '.join(sorted(_ALLOWED_SUBDIRS))}. Got: '{parts[0] if parts else ''}'",
@@ -330,7 +330,9 @@ class SkillFileTool(BaseTool):
target = skill_dir / rel_path
# Safety: prevent path traversal
try:
target.resolve().relative_to(skill_dir.resolve())
resolved_parts = target.resolve().relative_to(skill_dir.resolve()).parts
if len(resolved_parts) < 2 or resolved_parts[0] not in _ALLOWED_SUBDIRS:
raise ValueError("Path escapes allowed skill subdirectories")
except ValueError:
return json.dumps({"status": "error", "error": "Path escapes skill directory"})
@@ -356,7 +358,7 @@ class SkillFileTool(BaseTool):
# Validate subdirectory (same allowlist as _write_file, so remove can
# never touch a file write was never allowed to create)
parts = Path(rel_path).parts
if len(parts) < 2 or parts[0] not in _ALLOWED_SUBDIRS:
if len(parts) < 2 or parts[0] not in _ALLOWED_SUBDIRS or ".." in parts:
return json.dumps({
"status": "error",
"error": f"Path must start with one of: {', '.join(sorted(_ALLOWED_SUBDIRS))}. Got: '{parts[0] if parts else ''}'",
@@ -364,13 +366,17 @@ class SkillFileTool(BaseTool):
target = skill_dir / rel_path
try:
target.resolve().relative_to(skill_dir.resolve())
resolved_parts = target.resolve().relative_to(skill_dir.resolve()).parts
if len(resolved_parts) < 2 or resolved_parts[0] not in _ALLOWED_SUBDIRS:
raise ValueError("Path escapes allowed skill subdirectories")
except ValueError:
return json.dumps({"status": "error", "error": "Path escapes skill directory"})
if not target.exists():
return json.dumps({"status": "error", "error": f"File not found: {rel_path}"})
if not target.is_file():
return json.dumps({"status": "error", "error": "Only auxiliary files can be removed"})
target.unlink()
return json.dumps({
"status": "ok",
+13 -2
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import os
from pathlib import Path
import pytest
@@ -216,7 +215,7 @@ class TestCjkContent:
"币的 的买 买入 入和 和卖 卖出 出时 时机 。"
)
cleaned = MemorySearchIndex._clean_cjk(raw)
assert cleaned == "使用移动平均线来判断>>>比<<<>>>特<<<>>>币<<<的买入和卖出时机 。"
assert cleaned == "使用移动平均线来判断>>>比<<<>>>特<<<>>>币<<<的买入和卖出时机。"
def test_cjk_mixed_with_latin_round_trips_without_inserted_spaces(self) -> None:
"""A CJK run directly against a ticker, date, or punctuation mark
@@ -301,3 +300,15 @@ class TestGracefulDegradation:
# If it raises during init with a completely corrupted file,
# that's also acceptable - the key is no unhandled crash
pass
@pytest.mark.parametrize("original", ["研究AAPL价格。", "研究...价格", "研究 AAPL 价格。", "记忆 系统"])
def test_mixed_script_display_round_trip(original):
assert MemorySearchIndex._clean_cjk(MemorySearchIndex._prepare_cjk(original)) == original
def test_bigram_only_snippet_retains_highlights_and_truncation():
snippet = MemorySearchIndex._clean_cjk("...使用 >>>用移<<< 移动...")
assert snippet.startswith("...") and snippet.endswith("...")
assert ">>>" in snippet and "<<<" in snippet
assert snippet.replace(">>>", "").replace("<<<", "") == "...使用移动..."
+55
View File
@@ -475,3 +475,58 @@ def test_header_only_equity_csv_yields_empty_preview(summary_run_dir):
_write_equity_csv(summary_run_dir / "artifacts" / "equity.csv", [])
assert build_backtest_summary(summary_run_dir)["equity_preview"] == []
@pytest.mark.parametrize("relative", ["run_card.json", "artifacts/equity.csv"])
def test_summary_refuses_artifact_symlink_outside_run(summary_run_dir, tmp_path, relative):
path = summary_run_dir / relative
outside = tmp_path / "external"
outside.write_bytes(path.read_bytes())
path.unlink()
path.symlink_to(outside)
assert try_build_backtest_summary(summary_run_dir) is None
def test_ohlcv_paths_skip_external_symlinks_and_directories(summary_run_dir, tmp_path):
external = tmp_path / "external.csv"
external.write_text("timestamp,open\n2024-01-01,1\n")
(summary_run_dir / "artifacts/ohlcv_SECRET.csv").symlink_to(external)
(summary_run_dir / "artifacts/ohlcv_DIRECTORY.csv").mkdir()
assert set(collect_ohlcv_paths(summary_run_dir)) == {"BTC-USDT"}
def test_real_run_card_structured_metrics_reach_summary(summary_run_dir):
from backtest.run_card import write_run_card
card = write_run_card(summary_run_dir, {"codes": ["BTC-USDT"]}, {
"sharpe": 1.25, "unfilled_plan_rejections_by_symbol": {"BTC-USDT": 2},
"validation": {"passed": False}, "warnings": ["annualisation adjusted"],
})
summary = build_backtest_summary(summary_run_dir)
assert summary["metrics"] == card["metrics"]
assert summary["structured_metrics"] == card["structured_metrics"]
assert summary["validation"] == card["validation"]
def test_long_equity_summary_survives_model_delivery(summary_run_dir):
from src.config.limits import TOOL_RESULT_LIMIT, truncate_tool_result
_write_equity_csv(summary_run_dir / "artifacts/equity.csv", _equity_rows(10000))
result = _run_tool(summary_run_dir, _FakeRunResult(success=True, exit_code=0, stdout="x" * 4000, stderr="y" * 4000))
delivered = json.dumps(result, ensure_ascii=False)
assert len(delivered) <= TOOL_RESULT_LIMIT
assert truncate_tool_result(delivered) == delivered
assert result["summary"]["metrics"] == CARD_METRICS
preview = result["summary"]["equity_preview"]
assert preview[0]["equity"] == 1000000.0
assert preview[-1]["equity"] == 1000000.0 + 1000 * 9999
def test_oversized_metrics_summary_reports_omission(summary_run_dir):
from src.config.limits import TOOL_RESULT_LIMIT
card = _run_card()
card["structured_metrics"] = {"detail": "x" * TOOL_RESULT_LIMIT}
(summary_run_dir / "run_card.json").write_text(json.dumps(card))
result = _run_tool(summary_run_dir, _FakeRunResult(success=True, exit_code=0))
assert result["status"] == "ok"
assert "summary" not in result
assert "exceeds the result budget" in result["summary_omitted"]
assert len(json.dumps(result)) <= TOOL_RESULT_LIMIT
+20 -1
View File
@@ -203,6 +203,8 @@ class TestTruncationReporting:
"""A capped series reports the cap instead of looking complete."""
def _run(self, monkeypatch, count: int, **kwargs: Any) -> Dict[str, Any]:
# Isolate the requested row cap; the real delivery cap is tested below.
monkeypatch.setattr(fred_macro_tool, "TOOL_RESULT_LIMIT", 1_000_000)
monkeypatch.setenv("FRED_API_KEY", "tok_123")
monkeypatch.setattr(
fred_macro_tool,
@@ -303,9 +305,26 @@ class TestTruncationReporting:
envelope = FredMacroTool().execute(series_id="UNRATE")
delivered = truncate_tool_result(envelope)
assert len(envelope) > TOOL_RESULT_LIMIT
assert len(envelope) <= TOOL_RESULT_LIMIT
assert delivered == envelope
data = json.loads(delivered)["data"]
assert data["count"] == len(data["observations"])
assert data["truncated"] is True
assert "character budget" in data["hint"]
assert "raise limit" not in data["hint"]
assert data["observations"][-1]["value"] == 2499.0
assert len(delivered) <= TOOL_RESULT_LIMIT
for field in ("observations_available", "truncated", "limit", "hint"):
assert f'"{field}"' in delivered
# The observations are the part that gets cut.
assert delivered.count('"date"') < fred_macro_tool._DEFAULT_LIMIT
def test_upstream_partial_history_does_not_claim_complete(monkeypatch):
monkeypatch.setenv("FRED_API_KEY", "test")
monkeypatch.setattr(fred_macro_tool, "throttled_get_json", lambda *a, **kw: {
"count": 100001, "observations": [{"date": "2000-01-01", "value": "1"}]
})
result = json.loads(FredMacroTool().execute(series_id="TEST"))
assert result["ok"] is False
assert "narrow the date window" in result["error"]
+58 -5
View File
@@ -23,6 +23,7 @@ from src.tools.run_artifact_tool import RunArtifactTool, read_run_artifact
EQUITY_ROWS = 3821
EQUITY_COLUMNS = ["date", "equity", "cash", "drawdown"]
ROWS_ENVELOPE_KEYS = {
"path",
"artifact",
"run_dir",
"columns",
@@ -94,7 +95,8 @@ def _write_run_card(run_dir: Path, entries: list[dict]) -> None:
# --------------------------------------------------------------------------
def test_downsample_caps_points_and_pins_first_last(run_dir: Path) -> None:
def test_downsample_caps_points_and_pins_first_last(run_dir: Path, monkeypatch) -> None:
monkeypatch.setattr(run_artifact_tool, "_BYTE_BUDGET", 120_000)
env = json.loads(
read_run_artifact(str(run_dir), "equity", format="downsample", max_rows=1000)
)
@@ -378,7 +380,7 @@ def test_numeric_coercion(run_dir: Path) -> None:
def test_meta_format(run_dir: Path) -> None:
env = json.loads(read_run_artifact(str(run_dir), "equity", format="meta"))
assert set(env) == {"artifact", "run_dir", "columns", "total_rows", "size_bytes"}
assert set(env) == {"artifact", "run_dir", "path", "columns", "total_rows", "size_bytes"}
assert env["artifact"] == "equity"
assert env["columns"] == EQUITY_COLUMNS
assert env["total_rows"] == EQUITY_ROWS
@@ -463,9 +465,8 @@ def test_byte_budget_never_breaks_json_at_floor(
) -> None:
monkeypatch.setattr(run_artifact_tool, "_BYTE_BUDGET", 10)
env = json.loads(read_run_artifact(str(run_dir), "equity", max_rows=100))
assert env["returned_rows"] == 1
assert env["truncated"] is True
assert env["next_offset"] == 1
assert env["ok"] is False
assert "budget" in env["error"]
# --------------------------------------------------------------------------
@@ -563,3 +564,55 @@ def test_tool_class_contract(run_dir: Path) -> None:
)
assert set(default) == ROWS_ENVELOPE_KEYS
assert default["returned_rows"] == 3
def test_default_page_survives_actual_agent_delivery_cap(run_dir):
from src.config.limits import TOOL_RESULT_LIMIT, truncate_tool_result
payload = read_run_artifact(str(run_dir), "equity")
assert len(payload) <= TOOL_RESULT_LIMIT
assert truncate_tool_result(payload) == payload
first = json.loads(payload)
assert first["path"] == str(run_dir / "artifacts/equity.csv")
next_page = json.loads(read_run_artifact(str(run_dir), "equity", offset=first["next_offset"]))
assert next_page["rows"][0] == _expected_row(first["returned_rows"])
def test_oversized_json_is_an_actionable_error(run_dir):
from src.config.limits import TOOL_RESULT_LIMIT
(run_dir / "run_card.json").write_text(json.dumps({"body": "x" * TOOL_RESULT_LIMIT}))
payload = read_run_artifact(str(run_dir), "run_card")
assert len(payload) <= TOOL_RESULT_LIMIT
assert json.loads(payload)["ok"] is False
def test_oversized_row_requires_projection(run_dir):
_write_csv(run_dir / "artifacts/trades.csv", ["code", "note"], [["000001", "x" * 20000]])
assert json.loads(read_run_artifact(str(run_dir), "trades"))["ok"] is False
projected = json.loads(read_run_artifact(str(run_dir), "trades", columns=["code"]))
assert projected["rows"] == [["000001"]]
def test_shape_and_page_do_not_coerce_unselected_rows(run_dir, monkeypatch):
original = run_artifact_tool._coerce_value
count = 0
def coercion(value):
nonlocal count
count += 1
return original(value)
monkeypatch.setattr(run_artifact_tool, "_coerce_value", coercion)
read_run_artifact(str(run_dir), "equity", format="meta")
assert count == 0
read_run_artifact(str(run_dir), "equity", max_rows=3)
assert count == 3 * len(EQUITY_COLUMNS)
def test_oversized_json_meta_reports_size_without_content(run_dir):
from src.config.limits import TOOL_RESULT_LIMIT
card = run_dir / "run_card.json"
card.write_text(json.dumps({"body": "x" * TOOL_RESULT_LIMIT}))
result = json.loads(read_run_artifact(str(run_dir), "run_card", format="meta"))
assert result["size_bytes"] == card.stat().st_size
assert result["content_omitted"] is True
assert result["format"] == "json"
assert "json" not in result
+23 -1
View File
@@ -14,7 +14,6 @@ from src.tools.skill_writer_tool import (
DeleteSkillTool,
SkillFileTool,
_sanitize_skill_name,
USER_SKILLS_DIR,
)
@@ -312,3 +311,26 @@ class TestSkillCRUDLifecycle:
r = json.loads(delete.execute(name="lifecycle"))
assert r["status"] == "ok"
assert not (tmp_path / "lifecycle").exists()
@pytest.mark.parametrize("action", ["write", "remove"])
def test_auxiliary_operations_refuse_parent_traversal(tmp_path, action):
skill = tmp_path / "demo"
(skill / "assets").mkdir(parents=True)
(skill / "LICENSE").write_text("MIT")
with patch("src.tools.skill_writer_tool.USER_SKILLS_DIR", tmp_path):
result = json.loads(SkillFileTool().execute(action=action, skill_name="demo", path="assets/../LICENSE", content="poison"))
assert result["status"] == "error"
assert (skill / "LICENSE").read_text() == "MIT"
@pytest.mark.parametrize("action", ["write", "remove"])
def test_auxiliary_operations_refuse_symlink_into_forbidden_sibling(tmp_path, action):
skill = tmp_path / "demo"
(skill / "private").mkdir(parents=True)
(skill / "private/data").write_text("original")
(skill / "assets").symlink_to(skill / "private", target_is_directory=True)
with patch("src.tools.skill_writer_tool.USER_SKILLS_DIR", tmp_path):
result = json.loads(SkillFileTool().execute(action=action, skill_name="demo", path="assets/data", content="poison"))
assert result["status"] == "error"
assert (skill / "private/data").read_text() == "original"