mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
feat(acceptance): verify required CSV header columns (#6714)
* feat(acceptance): verify required CSV header columns * fix(acceptance): address CSV header review feedback * fix(acceptance): guard CSV criteria and malformed headers * test(acceptance): isolate CSV deadline clock simulations
This commit is contained in:
ZJPex
GitHub
parent
6783073a39
commit
e4c893f2fb
20
README.md
20
README.md
@ -2096,6 +2096,26 @@ For `tests_passed:go test ./...`, packages marked `[no test files]` or `[no test
|
||||
|
||||
For file acceptance criteria, an empty regular file in the shared workspace can satisfy `file:<path> exists` and `file_written:<path>`, including on remote sandboxes. It fails `file:<path> non-empty` with a deterministic empty-file result.
|
||||
|
||||
To require CSV header columns, set a `task` or `batch_task` item's criterion to
|
||||
`file:../outputs/report.csv csv-columns:["id","name"]`. The JSON array must contain
|
||||
unique, nonempty, UTF-8-encodable column names. Names match exactly (case and
|
||||
whitespace included), in any order; extra columns are allowed. Checks read only a bounded prefix and parse
|
||||
the first comma-separated logical record, supporting double-quoted fields, doubled
|
||||
quotes, embedded newlines, and an optional UTF-8 BOM. Data rows are not validated.
|
||||
Empty files, missing required columns, malformed or invalid UTF-8 headers, and
|
||||
empty or duplicate header names return `does not hold`. Headers over **50,000 bytes**, incomplete reads,
|
||||
invalid criteria, unavailable workspace access, or exhausted read budgets return
|
||||
`UNVERIFIED`. A large file can satisfy the check when its complete header fits the
|
||||
budget. A definite syntax error within that header remains a checked negative
|
||||
regardless of later data rows; truncated quoting remains `UNVERIFIED`.
|
||||
Reads use the same authorized, scoped transport as JSON checks below:
|
||||
50,001 bytes maximum; remote reads have a 5-second command timeout (10 seconds for
|
||||
the provider call). Local reads and parsing share a cooperative 5-second deadline;
|
||||
remote parsing has its own 5-second deadline starting after the read returns.
|
||||
These checks do not interrupt an in-progress filesystem or parser call. Concurrent
|
||||
writes are not an atomic snapshot. This does not validate data types, row counts,
|
||||
or business meaning.
|
||||
|
||||
To request JSON syntax validation, explicitly set a `task` or `batch_task` item's
|
||||
`acceptance_criteria` to `["file:../outputs/report.json json-valid"]`. Checks cover only
|
||||
complete UTF-8 JSON files within the shared workspace, up to **50,000 bytes**.
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
## Project Overview
|
||||
|
||||
The backend runs a LangGraph-based super agent with sandbox execution, persistent memory, subagent delegation, and extensible tools in isolated per-thread environments.
|
||||
The LangGraph backend provides sandbox execution, persistent memory, subagent delegation, and extensible tools, isolated per thread.
|
||||
|
||||
**Architecture**:
|
||||
- **Gateway API** (port 8001): REST API plus embedded LangGraph-compatible agent runtime
|
||||
@ -13,7 +13,7 @@ The backend runs a LangGraph-based super agent with sandbox execution, persisten
|
||||
**Runtime**:
|
||||
- `make dev`, Docker dev, and production all run the agent runtime in Gateway via `RunManager` + `run_agent()` + `StreamBridge` (`packages/harness/deerflow/runtime/`). Nginx exposes that runtime at `/api/langgraph/*` and rewrites it to Gateway's native `/api/*` routers.
|
||||
- Gateway streams `write_file` and `str_replace` argument deltas in bounded batches for multi-mode `messages-tuple` consumers; single-mode message consumers retain the original per-chunk contract. Non-message frames flush pending batches, and `values` remains an optional complete-state snapshot rather than a prerequisite for batching.
|
||||
- With `stream_subgraphs`, subgraph frames keep their namespace in the SSE event name (`values|<ns>`, LangGraph Platform style) instead of impersonating root frames — a delegated subagent inherits the parent checkpoint namespace, so publishing its `values` snapshot as bare `values` replaces the whole thread view in SDK clients (#4399). Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend does not request subgraph streaming; subtask progress rides root-namespace `task_*` custom events.
|
||||
- `stream_subgraphs` preserves subgraph SSE namespaces (`values|<ns>`, LangGraph Platform style); bare `values` would overwrite the SDK thread view (#4399). Delegated subagents inherit the parent's checkpoint namespace. Root-only consumers (file-tool chunk batcher, subagent event persistence, LLM error-fallback detection) ignore namespaced frames. The web frontend uses root `task_*` events without subgraph streaming.
|
||||
- Background subagent identity is deliberately split: the provider `tool_call_id` remains the correlation key for `ToolMessage`, `task_*` SSE events, persisted lifecycle events, frontend cards, and the public `ExtensionData.scope_id` contract (stored as `SubagentResult.external_task_id`), while `SubagentExecutor.execute_async()` generates a full server-side `execution_id` for `SubagentResult.task_id`, the process-wide registry, polling, cancellation, timeout handling, and cleanup. Provider IDs are not globally unique across parent runs, so they must never become registry ownership keys; scheduler closures retain their own `SubagentResult` rather than resolving ownership again through the mutable registry. Terminal subagent token usage travels in the current run's `ToolMessage.additional_kwargs` and is attributed from message state, never through a process-global provider-ID cache.
|
||||
- Scheduled tasks dispatch through the normal Gateway run path. `launch_scheduled_thread_run` reads `get_app_config()` at dispatch and passes `scheduler.recursion_limit` (default 1000, matching the web UI; clamped by `max_recursion_limit`), so YAML changes apply on the next run without restarting Gateway.
|
||||
- Run-history `status` filters are occurrence states, not task states. `ScheduledTaskRunStatus` in `persistence/scheduled_tasks/model.py` is the shared API/repository vocabulary and must match the active and terminal occurrence-status sets. Keep owner lookup before reading history, and apply SQL task/status predicates before pagination; omitted status preserves the existing response.
|
||||
|
||||
@ -4,6 +4,8 @@ Result-reader acceptance runs on SQLite and, with TEST_POSTGRES_URI, an isolated
|
||||
PostgreSQL schema; both exercise submission, completion, reopen and ToolNode
|
||||
continuation without paid model work. Always clean up the temporary schema.
|
||||
|
||||
**CSV headers**: `file:<path> csv-columns:["id","name"]` uses shared file admission and authorized bounded reads. Require UTF-8-encodable decoded column names before file access. Parse only the first logical record; definite syntax errors within the header budget remain checked negatives regardless of later rows, while truncated quoting and exhausted budgets stay UNVERIFIED. Local reads and parsing share a cooperative 5-second deadline; remote transport keeps its own timeouts and parsing starts a separate 5-second deadline after the read returns. See README for exact matching and dialect.
|
||||
|
||||
**JSON**: See README. Authorize before local metadata and reads; inaccessible paths stay UNVERIFIED.
|
||||
|
||||
Apply each subagent's prompt overlay after assembling its full SystemMessage.
|
||||
|
||||
@ -8,6 +8,9 @@ requirement.
|
||||
|
||||
Leaf families:
|
||||
|
||||
- ``file:<path> csv-columns:["id","name"]`` — UTF-8 CSV header only, optional BOM,
|
||||
exact required columns. Bounded prefix; no data-row or schema validation.
|
||||
|
||||
- ``file:<path> json-valid`` — explicit UTF-8 JSON syntax check, capped at 50,000 bytes.
|
||||
Read up to the cap plus one byte and verify completeness; oversize, truncated reads
|
||||
and parser resource limits remain UNVERIFIED. Reject NaN/Infinity; no schema or semantic validation.
|
||||
@ -60,6 +63,8 @@ from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import csv
|
||||
import io
|
||||
import json
|
||||
import ntpath
|
||||
import os
|
||||
@ -67,6 +72,7 @@ import posixpath
|
||||
import re
|
||||
import shlex
|
||||
import stat
|
||||
import time
|
||||
import unicodedata
|
||||
from collections.abc import Callable, Mapping
|
||||
from typing import Any, TypedDict
|
||||
@ -93,6 +99,7 @@ _PROVIDER_ERROR_PREFIX = "Error:"
|
||||
|
||||
_FILE_LEAF_RE = re.compile(r"^file:(?P<path>.+?)\s+(?P<mode>exists|non-empty)$", re.IGNORECASE)
|
||||
_JSON_FILE_LEAF_RE = re.compile(r"^file:(?P<path>.+?)\s+json-valid$", re.IGNORECASE)
|
||||
_CSV_FILE_LEAF_RE = re.compile(r"^file:(?P<path>.+?)\s+csv-columns:(?P<columns>.+)$", re.IGNORECASE)
|
||||
_FILE_WRITTEN_RE = re.compile(r"^file_written:(?P<path>.+)$", re.IGNORECASE)
|
||||
_TESTS_PASSED_RE = re.compile(r"^tests_passed:(?P<command>.+)$", re.IGNORECASE)
|
||||
|
||||
@ -137,7 +144,7 @@ _TEST_FAIL_SHAPE_RE = re.compile(
|
||||
|
||||
class AcceptanceLeaf(TypedDict):
|
||||
criterion: str # original criterion text (bounded)
|
||||
family: str # file_exists | file_non_empty | file_json_valid | file_written | tests_passed | undecidable
|
||||
family: str # file_exists | file_non_empty | file_json_valid | file_csv_columns | file_written | tests_passed | undecidable
|
||||
checked: bool # a deterministic check ran
|
||||
holds: bool # checked AND the condition holds; always False when unchecked
|
||||
detail: str # short evidence note (bounded)
|
||||
@ -164,6 +171,9 @@ def parse_file_criterion(criterion: str) -> tuple[str, str] | None:
|
||||
json_match = _JSON_FILE_LEAF_RE.match(criterion)
|
||||
if json_match is not None:
|
||||
return "file_json_valid", json_match.group("path")
|
||||
csv_match = _CSV_FILE_LEAF_RE.match(criterion)
|
||||
if csv_match is not None:
|
||||
return "file_csv_columns", csv_match.group("path")
|
||||
written_match = _FILE_WRITTEN_RE.match(criterion)
|
||||
if written_match is not None:
|
||||
return "file_written", written_match.group("path")
|
||||
@ -420,7 +430,7 @@ _JSON_READ_INNER_SCRIPT = (
|
||||
)
|
||||
|
||||
|
||||
def _read_bounded_json_content(runtime: Any, resolved: str, thread_data: Mapping[str, Any] | None) -> bytes | None:
|
||||
def _read_bounded_file_content(runtime: Any, resolved: str, thread_data: Mapping[str, Any] | None, *, deadline: float | None = None) -> bytes | None:
|
||||
"""Read up to the cap plus one byte to detect growth; return None if completeness is uncertain."""
|
||||
from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
|
||||
from deerflow.sandbox.tools import _resolve_local_read_path, ensure_sandbox_initialized, is_local_sandbox
|
||||
@ -460,10 +470,14 @@ def _read_bounded_json_content(runtime: Any, resolved: str, thread_data: Mapping
|
||||
content = bytearray()
|
||||
# Disable buffered read-ahead; continue short reads to EOF or cap plus one to distinguish complete from truncated content.
|
||||
while len(content) <= _FILE_CONTENT_READ_CAP_BYTES:
|
||||
if deadline is not None and time.monotonic() >= deadline:
|
||||
return None
|
||||
chunk = handle.read(_FILE_CONTENT_READ_CAP_BYTES + 1 - len(content))
|
||||
if not chunk:
|
||||
break
|
||||
content.extend(chunk)
|
||||
if deadline is not None and time.monotonic() >= deadline:
|
||||
return None
|
||||
return bytes(content)
|
||||
|
||||
|
||||
@ -483,7 +497,7 @@ def _check_json_file(base: AcceptanceLeaf, runtime: Any, resolved: str, thread_d
|
||||
base["detail"] = f"JSON file exceeds {_FILE_CONTENT_READ_CAP_BYTES}-byte read cap; content not read"
|
||||
return base
|
||||
try:
|
||||
content = _read_bounded_json_content(runtime, resolved, thread_data)
|
||||
content = _read_bounded_file_content(runtime, resolved, thread_data)
|
||||
except (FileNotFoundError, SandboxFileNotFoundError):
|
||||
base["checked"] = True
|
||||
base["detail"] = "file does not exist"
|
||||
@ -510,6 +524,103 @@ def _check_json_file(base: AcceptanceLeaf, runtime: Any, resolved: str, thread_d
|
||||
return base
|
||||
|
||||
|
||||
def _parse_csv_columns(criterion: str) -> list[str] | None:
|
||||
"""Require an explicit JSON array of unique, nonempty, exact column names."""
|
||||
match = _CSV_FILE_LEAF_RE.match(criterion)
|
||||
if match is None:
|
||||
return None
|
||||
try:
|
||||
columns = json.loads(match.group("columns"))
|
||||
except (ValueError, RecursionError):
|
||||
return None
|
||||
if not isinstance(columns, list) or not columns or any(not isinstance(column, str) or not column for column in columns):
|
||||
return None
|
||||
try:
|
||||
for column in columns:
|
||||
column.encode("utf-8")
|
||||
except UnicodeEncodeError:
|
||||
return None
|
||||
return columns if len(set(columns)) == len(columns) else None
|
||||
|
||||
|
||||
def _check_csv_header(base: AcceptanceLeaf, runtime: Any, resolved: str, thread_data: Mapping[str, Any] | None, probed_size: int, columns: list[str]) -> AcceptanceLeaf:
|
||||
"""Check only the first CSV record from a bounded prefix, never the data rows."""
|
||||
from deerflow.sandbox.exceptions import SandboxError, SandboxFileNotFoundError
|
||||
from deerflow.sandbox.tools import is_local_sandbox
|
||||
|
||||
deadline = time.monotonic() + 5 if is_local_sandbox(runtime) else None
|
||||
try:
|
||||
content = _read_bounded_file_content(runtime, resolved, thread_data, deadline=deadline)
|
||||
except (FileNotFoundError, SandboxFileNotFoundError):
|
||||
base["checked"] = True
|
||||
base["detail"] = "file does not exist"
|
||||
return base
|
||||
except (OSError, SandboxError):
|
||||
base["detail"] = "bounded CSV read failed"
|
||||
return base
|
||||
# Remote transport owns its timeouts; start the parser budget after it returns.
|
||||
if deadline is None:
|
||||
deadline = time.monotonic() + 5
|
||||
if time.monotonic() >= deadline:
|
||||
base["detail"] = "CSV header deadline exceeded"
|
||||
return base
|
||||
if content is None or len(content) != min(probed_size, _FILE_CONTENT_READ_CAP_BYTES + 1):
|
||||
base["detail"] = "complete CSV prefix unavailable within the read cap"
|
||||
return base
|
||||
# Surrogate escape lets an invalid or split UTF-8 sequence in a data row
|
||||
# remain irrelevant. Validate UTF-8 only in the consumed header below.
|
||||
text = content.decode("utf-8-sig", errors="surrogateescape")
|
||||
stream = io.StringIO(text, newline="")
|
||||
complete = len(content) == probed_size
|
||||
malformed = False
|
||||
try:
|
||||
header = next(csv.reader(stream, strict=True), [])
|
||||
except csv.Error as exc:
|
||||
if "field larger than field limit" in str(exc):
|
||||
base["detail"] = "CSV parser resource limit; header unverified"
|
||||
return base
|
||||
if str(exc) == "unexpected end of data" and not complete:
|
||||
base["detail"] = "CSV header incomplete within the read cap"
|
||||
return base
|
||||
# A syntax error already observed in the header cannot be repaired by later data rows.
|
||||
malformed = True
|
||||
header = []
|
||||
except MemoryError:
|
||||
base["detail"] = "CSV parser resource limit; header unverified"
|
||||
return base
|
||||
consumed = text[: stream.tell()]
|
||||
header_bytes = len(consumed.encode("utf-8", errors="surrogateescape")) + (3 if content.startswith(b"\xef\xbb\xbf") else 0)
|
||||
if not malformed and not complete and not consumed.endswith(("\r", "\n")):
|
||||
base["detail"] = "CSV header incomplete within the read cap"
|
||||
return base
|
||||
if header_bytes > _FILE_CONTENT_READ_CAP_BYTES:
|
||||
base["detail"] = f"CSV header exceeds the {_FILE_CONTENT_READ_CAP_BYTES}-byte read budget"
|
||||
return base
|
||||
if time.monotonic() >= deadline:
|
||||
base["detail"] = "CSV header deadline exceeded"
|
||||
return base
|
||||
if malformed:
|
||||
base["checked"] = True
|
||||
base["detail"] = "invalid CSV header"
|
||||
return base
|
||||
try:
|
||||
consumed.encode("utf-8")
|
||||
except UnicodeEncodeError:
|
||||
base["checked"] = True
|
||||
base["detail"] = "CSV header is not UTF-8"
|
||||
return base
|
||||
base["checked"] = True
|
||||
if not header or any(not column for column in header) or len(set(header)) != len(header):
|
||||
base["detail"] = "CSV header is empty or contains empty or duplicate column names"
|
||||
elif not set(columns).issubset(header):
|
||||
missing = sorted(set(columns) - set(header))
|
||||
base["detail"] = f"CSV header is missing required columns: {json.dumps(missing, ensure_ascii=False)}"
|
||||
else:
|
||||
base["holds"] = True
|
||||
base["detail"] = "CSV header contains required columns; data rows not validated"
|
||||
return base
|
||||
|
||||
|
||||
def _check_file_leaf(
|
||||
family: str,
|
||||
path: str,
|
||||
@ -519,6 +630,7 @@ def _check_file_leaf(
|
||||
content_reader: Callable[[Any, str], str],
|
||||
size_prober: Callable[[Any, str, Mapping[str, Any] | None], int | None],
|
||||
readable_prober: Callable[[Any, str, Mapping[str, Any] | None], bool | None],
|
||||
csv_columns: list[str] | None = None,
|
||||
) -> AcceptanceLeaf:
|
||||
criterion_path = path.strip()
|
||||
# Lazy imports: the sandbox helpers pull the provider stack, and this
|
||||
@ -528,14 +640,17 @@ def _check_file_leaf(
|
||||
from deerflow.sandbox.tools import is_local_sandbox
|
||||
|
||||
base: AcceptanceLeaf = {"criterion": "", "family": family, "checked": False, "holds": False, "detail": ""}
|
||||
if family == "file_json_valid" and is_local_sandbox(runtime):
|
||||
if family == "file_csv_columns" and csv_columns is None:
|
||||
base["detail"] = "csv-columns requires a nonempty JSON array of unique nonempty strings"
|
||||
return base
|
||||
if family in ("file_json_valid", "file_csv_columns") and is_local_sandbox(runtime):
|
||||
from deerflow.authz.sandbox_authz import authorize_sandbox_execution, safe_app_config
|
||||
|
||||
# Authorize before canonicalization or metadata probes can disclose file existence.
|
||||
try:
|
||||
authorize_sandbox_execution(context=getattr(runtime, "context", None) or {}, app_config=safe_app_config())
|
||||
except SandboxError:
|
||||
base["detail"] = "sandbox access denied; JSON file unverified"
|
||||
base["detail"] = "sandbox access denied; file unverified"
|
||||
return base
|
||||
|
||||
# Symlink escapes are a local-sandbox concern (host-visible links); remote
|
||||
@ -558,6 +673,8 @@ def _check_file_leaf(
|
||||
# worker — degrade to UNVERIFIED instead.
|
||||
base["detail"] = "file size could not be established by a bounded probe; content not read"
|
||||
return base
|
||||
if family == "file_csv_columns" and csv_columns is not None:
|
||||
return _check_csv_header(base, runtime, resolved, thread_data, probed_size, csv_columns)
|
||||
if family == "file_json_valid":
|
||||
return _check_json_file(base, runtime, resolved, thread_data, probed_size)
|
||||
if probed_size > _FILE_CONTENT_READ_CAP_BYTES:
|
||||
@ -1876,7 +1993,16 @@ def check_acceptance_criteria(
|
||||
tests_match = _TESTS_PASSED_RE.match(criterion)
|
||||
if file_criterion is not None:
|
||||
family, path = file_criterion
|
||||
leaf = _check_file_leaf(family, path, runtime=runtime, thread_data=thread_data, content_reader=content_reader, size_prober=size_prober, readable_prober=readable_prober)
|
||||
leaf = _check_file_leaf(
|
||||
family,
|
||||
path,
|
||||
runtime=runtime,
|
||||
thread_data=thread_data,
|
||||
content_reader=content_reader,
|
||||
size_prober=size_prober,
|
||||
readable_prober=readable_prober,
|
||||
csv_columns=_parse_csv_columns(criterion) if family == "file_csv_columns" else None,
|
||||
)
|
||||
elif tests_match is not None:
|
||||
leaf = _check_tests_passed_leaf(tests_match.group("command"), bash_executions, thread_data)
|
||||
else:
|
||||
|
||||
@ -742,6 +742,10 @@ async def task_tool(
|
||||
syntax up to 50,000 bytes, rejecting NaN/Infinity. Oversize files, incomplete reads,
|
||||
or parser resource limits return UNVERIFIED. No schema or business validation;
|
||||
.json files are not checked automatically.
|
||||
- CSV deliverables can opt into `file:<path> csv-columns:["id","name"]`: check the
|
||||
first UTF-8 header record (optional BOM), with exact required names and extra columns
|
||||
allowed. Empty/duplicate headers fail; incomplete or oversized headers are UNVERIFIED.
|
||||
The header budget is 50,000 bytes; data rows are not validated.
|
||||
- `completed` means execution ended, not task acceptance. Read each criterion
|
||||
and retain useful work. For `does not hold`, inspect the reason and repair
|
||||
or recheck only the unmet condition, reusing unaffected outputs.
|
||||
|
||||
@ -171,6 +171,41 @@ async def test_json_criterion_survives_batch_execution_and_storage(env):
|
||||
assert item["attempt"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_csv_header_criterion_survives_batch_execution_and_storage(env):
|
||||
"""CSV criteria use the same authorized batch lease and persisted verdict."""
|
||||
output = env.paths.sandbox_outputs_dir("thread-1", user_id="user-1") / "report.csv"
|
||||
output.write_bytes(b"id,other\n")
|
||||
batch = await _submit(env, ['file:../outputs/report.csv csv-columns:["id","name"]'])
|
||||
await _execute(env)
|
||||
item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0]
|
||||
assert item["status"] == "succeeded"
|
||||
leaf = item["acceptance_verdict"]["leaves"][0]
|
||||
assert (leaf["family"], leaf["checked"], leaf["holds"]) == ("file_csv_columns", True, False)
|
||||
assert item["attempt"] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("column", [r"\ud800", r"\udfff", r"id\ud800"])
|
||||
async def test_surrogate_csv_criterion_preserves_completed_execution_after_reopen(env, column):
|
||||
output = env.paths.sandbox_outputs_dir("thread-1", user_id="user-1") / "report.csv"
|
||||
output.write_bytes(b"id,name\n")
|
||||
criterion = f'file:../outputs/report.csv csv-columns:["{column}"]'
|
||||
batch = await _submit(env, [criterion])
|
||||
await _execute(env)
|
||||
repo = SubagentBatchRepository(get_session_factory())
|
||||
item = (await repo.list_items(batch["id"], user_id="user-1"))[0]
|
||||
assert item["status"] == "succeeded"
|
||||
assert item["attempt"] == 1
|
||||
assert item["acceptance_criteria"] == [criterion]
|
||||
assert item["result_preview"] == env.result.result
|
||||
leaf = item["acceptance_verdict"]["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
||||
json.dumps(item["acceptance_verdict"], ensure_ascii=False).encode("utf-8")
|
||||
assert (await repo.get_batch(batch["id"], user_id="user-1"))["status"] == "completed"
|
||||
assert await repo.claim_items(now=datetime.now(UTC) + timedelta(minutes=5), lease_owner="other", lease_seconds=60, limit=1) == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_criteria_skips_checker_and_preserves_success(env, monkeypatch):
|
||||
checker = AsyncMock(side_effect=AssertionError("no checklist"))
|
||||
|
||||
343
backend/tests/test_csv_acceptance.py
Normal file
343
backend/tests/test_csv_acceptance.py
Normal file
@ -0,0 +1,343 @@
|
||||
"""CSV header acceptance through the public, workspace-scoped checklist."""
|
||||
|
||||
import json
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import Mock
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.subagents import acceptance_checks as checks
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def parser_clock(monkeypatch):
|
||||
import time
|
||||
|
||||
original_monotonic = time.monotonic
|
||||
clock = Mock(return_value=0)
|
||||
monkeypatch.setattr(checks, "time", SimpleNamespace(monotonic=clock))
|
||||
yield clock
|
||||
# Deadline simulations must leave event loops and other threads on the real clock.
|
||||
assert time.monotonic is original_monotonic
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def check_csv(tmp_path):
|
||||
def check(content, columns=None, path="report.csv", **kwargs):
|
||||
if content is not None:
|
||||
(tmp_path / "report.csv").write_bytes(content)
|
||||
criterion = f"file:{path} csv-columns:{json.dumps(columns or ['id', 'name'])}"
|
||||
return checks.check_acceptance_criteria(
|
||||
[criterion],
|
||||
runtime=SimpleNamespace(state={"sandbox": {"sandbox_id": "local"}}),
|
||||
thread_data={"workspace_path": str(tmp_path)},
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
return check
|
||||
|
||||
|
||||
@pytest.mark.parametrize("content", [b"id,name\n1,A", b"\xef\xbb\xbfid,name\r\n", b'"id","name",extra\n', b"id,name", b"id,name\n" + b"\xff" * 60_000], ids=["plain", "bom", "quoted", "no-newline", "large-invalid-body"])
|
||||
def test_header_holds_without_validating_body(check_csv, content):
|
||||
assert check_csv(content)["all_hold"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("content", [b"", b"\n", b"\xef\xbb\xbf", b"id,other\n", b"id,name,name\n", b"id,name,\n", b"ID,name\n", b"id, name\n", b'"id","name\n', b"id,\xff\n"])
|
||||
def test_invalid_or_missing_header_does_not_hold(check_csv, content):
|
||||
leaf = check_csv(content)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
||||
|
||||
|
||||
def test_quoted_columns_and_multiline_header(check_csv):
|
||||
assert check_csv(b'"last,name","line\nbreak","a""b"\r\n', ["last,name", "line\nbreak", 'a"b'])["all_hold"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size,checked", [(50_000, True), (50_001, False)])
|
||||
def test_header_byte_limit(check_csv, size, checked):
|
||||
leaf = check_csv(b"id,name," + b"x" * (size - 9) + b"\n" + b"1" * 60_000)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (checked, checked)
|
||||
|
||||
|
||||
def test_partial_header_cannot_hold(check_csv):
|
||||
leaf = check_csv(b"id,name", size_prober=lambda *args: 100)["leaves"][0]
|
||||
assert leaf["checked"] is False
|
||||
|
||||
|
||||
def test_missing_file_is_negative(check_csv):
|
||||
leaf = check_csv(None)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("path", ["../outside.csv", "/mnt/user-data/uploads/report.csv"])
|
||||
def test_outside_shared_workspace_never_reads(check_csv, path):
|
||||
probe = Mock(side_effect=AssertionError("out of scope"))
|
||||
assert check_csv(b"id,name\n", path=path, size_prober=probe)["leaves"][0]["checked"] is False
|
||||
probe.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("payload", ["[]", '["id","id"]', "[1]", "null", '[""]', '["id",]', "id,name"])
|
||||
def test_invalid_criteria_are_unverified(payload):
|
||||
criterion = f"file:report.csv csv-columns:{payload}"
|
||||
verdict = checks.check_acceptance_criteria([criterion])
|
||||
assert verdict["unchecked"] == [criterion]
|
||||
assert verdict["all_hold"] is False
|
||||
|
||||
|
||||
def test_admission_uses_csv_path():
|
||||
assert checks.parse_file_criterion('file:report.csv csv-columns:["id"]') == ("file_csv_columns", "report.csv")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("column", ["\ud800", "\udfff", "id\ud800"])
|
||||
def test_surrogate_columns_are_invalid_before_file_access(check_csv, column):
|
||||
probe = Mock(side_effect=AssertionError("invalid criteria must not access files"))
|
||||
verdict = check_csv(b"id,name\n", [column], size_prober=probe)
|
||||
assert verdict["leaves"][0]["checked"] is False
|
||||
json.dumps(verdict, ensure_ascii=False).encode("utf-8")
|
||||
probe.assert_not_called()
|
||||
|
||||
|
||||
def test_escaped_surrogate_pair_is_a_valid_unicode_column(check_csv):
|
||||
assert check_csv("id,\U0001f600\n".encode(), ["\U0001f600"])["all_hold"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("header", [b'"id"x,name\n', b'"line\nbreak"x,name\r\n', b'"id"x,name\r'])
|
||||
@pytest.mark.parametrize("large_body", [False, True])
|
||||
def test_definite_malformed_header_is_negative_regardless_of_body(check_csv, header, large_body):
|
||||
leaf = check_csv(header + (b"x" * 60_000 if large_body else b""))["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
||||
assert leaf["detail"] == "invalid CSV header"
|
||||
|
||||
|
||||
def test_truncated_quoted_header_remains_unverified(check_csv):
|
||||
leaf = check_csv(b'"id\n' + b"x" * 60_000)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
||||
assert leaf["detail"] == "CSV header incomplete within the read cap"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("size,checked", [(50_000, True), (50_001, False)])
|
||||
def test_malformed_header_respects_byte_budget(check_csv, size, checked):
|
||||
leaf = check_csv(b'"id"x,' + b"x" * (size - 7) + b"\n" + b"x" * 60_000)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (checked, False)
|
||||
|
||||
|
||||
def test_malformed_header_respects_parser_deadline(check_csv, monkeypatch, parser_clock):
|
||||
reader = checks.csv.reader
|
||||
|
||||
def delayed_parse(*args, **kwargs):
|
||||
parser_clock.return_value = 6
|
||||
return reader(*args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(checks.csv, "reader", delayed_parse)
|
||||
leaf = check_csv(b'"id"x,name\n')["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
||||
assert leaf["detail"] == "CSV header deadline exceeded"
|
||||
|
||||
|
||||
def test_revoked_access_prevents_metadata(check_csv, monkeypatch):
|
||||
from deerflow.sandbox.exceptions import SandboxError
|
||||
|
||||
monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution", Mock(side_effect=SandboxError("denied")))
|
||||
probe = Mock(side_effect=AssertionError("must authorize first"))
|
||||
assert check_csv(b"id,name\n", size_prober=probe)["leaves"][0]["checked"] is False
|
||||
probe.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("content", [b'id,name\n"unterminated', b"id,name\r1,A\r", b"id,name\r\n1,A\r\n"])
|
||||
def test_only_first_logical_record_is_checked(check_csv, content):
|
||||
assert check_csv(content)["all_hold"] is True
|
||||
|
||||
|
||||
def test_unknown_size_does_not_read(check_csv, monkeypatch):
|
||||
reader = Mock(side_effect=AssertionError("no size evidence"))
|
||||
monkeypatch.setattr(checks, "_read_bounded_file_content", reader)
|
||||
assert check_csv(b"id,name\n", size_prober=lambda *args: None)["leaves"][0]["checked"] is False
|
||||
reader.assert_not_called()
|
||||
|
||||
|
||||
def test_expired_read_is_unverified(check_csv, monkeypatch, parser_clock):
|
||||
def delayed(*args, **kwargs):
|
||||
parser_clock.return_value = 6
|
||||
return b"id,name\n"
|
||||
|
||||
monkeypatch.setattr(checks, "_read_bounded_file_content", delayed)
|
||||
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", [None, OSError("unreadable")])
|
||||
def test_read_failure_remains_unverified(check_csv, monkeypatch, failure):
|
||||
reader = Mock(side_effect=failure) if failure else Mock(return_value=None)
|
||||
monkeypatch.setattr(checks, "_read_bounded_file_content", reader)
|
||||
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
def test_remote_transport_checks_deadline_and_completion(monkeypatch, tmp_path):
|
||||
import base64
|
||||
|
||||
payload = "JSON\n" + base64.b64encode(b"id,name\n\n000").decode() + "\nEND\n"
|
||||
sandbox = Mock()
|
||||
sandbox.execute_command.return_value = payload
|
||||
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
|
||||
|
||||
def check():
|
||||
return checks.check_acceptance_criteria(
|
||||
['file:report.csv csv-columns:["id","name"]'],
|
||||
runtime=SimpleNamespace(state=None),
|
||||
thread_data={"workspace_path": str(tmp_path)},
|
||||
size_prober=lambda *args: 8,
|
||||
)
|
||||
|
||||
assert check()["all_hold"] is True
|
||||
command = sandbox.execute_command.call_args
|
||||
assert command.kwargs["timeout"] == 10
|
||||
assert "/usr/bin/timeout 5" in command.args[0]
|
||||
assert command.args[0].endswith("50001")
|
||||
sandbox.execute_command.return_value = payload[:-5]
|
||||
assert check()["leaves"][0]["checked"] is False
|
||||
sandbox.execute_command.return_value = "JSON\n" + base64.b64encode(b"id,name\n\n124").decode() + "\nEND\n"
|
||||
assert check()["leaves"][0]["checked"] is False
|
||||
sandbox.read_file.assert_not_called()
|
||||
|
||||
|
||||
def test_parser_resource_limit_is_unverified(check_csv, monkeypatch):
|
||||
monkeypatch.setattr(checks.csv, "reader", Mock(side_effect=checks.csv.Error("field larger than field limit (2)")))
|
||||
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
def test_missing_workspace_is_unverified():
|
||||
verdict = checks.check_acceptance_criteria(['file:report.csv csv-columns:["id"]'])
|
||||
assert verdict["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
def test_directory_is_unverified(check_csv, tmp_path):
|
||||
(tmp_path / "report.csv").mkdir()
|
||||
assert check_csv(None)["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
def test_local_read_rechecks_authorization(check_csv, monkeypatch):
|
||||
from deerflow.sandbox.exceptions import SandboxError
|
||||
|
||||
authorize = Mock(side_effect=[None, SandboxError("revoked")])
|
||||
monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution", authorize)
|
||||
assert check_csv(b"id,name\n")["leaves"][0]["checked"] is False
|
||||
assert authorize.call_count == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"content,columns,checked,holds",
|
||||
[
|
||||
(b"\xef\xbb\xbfid,name\r\n", ["id", "name"], True, True),
|
||||
(b'"last,name","line\nbreak"\n', ["last,name", "line\nbreak"], True, True),
|
||||
(b"id,name,name\n", ["id"], True, False),
|
||||
(b"", ["id"], True, False),
|
||||
(b"id,name\n" + b"x" * 60_000, ["id", "name"], True, True),
|
||||
(b'"' + b"x" * 50_001, ["id"], False, False),
|
||||
(b'"id"x,name\n' + b"x" * 60_000, ["id"], True, False),
|
||||
],
|
||||
ids=["bom", "multiline", "duplicate", "empty", "large-body", "oversized-header", "malformed-header-large-body"],
|
||||
)
|
||||
def test_remote_csv_uses_real_bounded_transport(tmp_path, monkeypatch, content, columns, checked, holds):
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
if sys.platform != "linux":
|
||||
pytest.skip("Requires Linux sandbox command tools")
|
||||
(tmp_path / "report.csv").write_bytes(content)
|
||||
|
||||
class ShellSandbox:
|
||||
def execute_command(self, command, **kwargs):
|
||||
assert kwargs["env"]
|
||||
args = shlex.split(command)
|
||||
prefix = "/mnt/user-data/workspace"
|
||||
args = [str(tmp_path) + arg[len(prefix) :] if arg == prefix or arg.startswith(prefix + "/") else arg for arg in args]
|
||||
return subprocess.run(args, capture_output=True, text=True, check=True, timeout=15).stdout
|
||||
|
||||
def read_file(self, *args, **kwargs):
|
||||
pytest.fail("CSV acceptance must not fall back to unbounded reads")
|
||||
|
||||
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: ShellSandbox())
|
||||
verdict = checks.check_acceptance_criteria(
|
||||
[f"file:report.csv csv-columns:{json.dumps(columns)}"],
|
||||
runtime=SimpleNamespace(state=None),
|
||||
thread_data={"workspace_path": str(tmp_path)},
|
||||
)
|
||||
leaf = verdict["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (checked, holds)
|
||||
|
||||
|
||||
def test_header_budget_counts_utf8_bytes(check_csv):
|
||||
verdict = check_csv(("id,name," + "\u4e2d" * 16_665 + "\n").encode())
|
||||
assert verdict["leaves"][0]["checked"] is False
|
||||
|
||||
|
||||
def test_header_without_newline_at_exact_cap_holds(check_csv):
|
||||
assert check_csv(b"id,name," + b"x" * (50_000 - 8))["all_hold"] is True
|
||||
|
||||
|
||||
def test_required_columns_allow_reordering(check_csv):
|
||||
assert check_csv(b"extra,name,id\n")["all_hold"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("parse_seconds,holds", [(0, True), (6, False)])
|
||||
def test_remote_read_latency_does_not_consume_parser_deadline(monkeypatch, tmp_path, parse_seconds, holds, parser_clock):
|
||||
import base64
|
||||
|
||||
sandbox = Mock()
|
||||
|
||||
def delayed_read(*args, **kwargs):
|
||||
parser_clock.return_value = 6
|
||||
return "JSON\n" + base64.b64encode(b"id,name\n\n000").decode() + "\nEND\n"
|
||||
|
||||
sandbox.execute_command.side_effect = delayed_read
|
||||
monkeypatch.setattr("deerflow.sandbox.tools.ensure_sandbox_initialized", lambda runtime: sandbox)
|
||||
reader = checks.csv.reader
|
||||
|
||||
def delayed_parse(*args, **kwargs):
|
||||
header = next(reader(*args, **kwargs))
|
||||
parser_clock.return_value += parse_seconds
|
||||
return iter([header])
|
||||
|
||||
monkeypatch.setattr(checks.csv, "reader", delayed_parse)
|
||||
verdict = checks.check_acceptance_criteria(
|
||||
['file:report.csv csv-columns:["id","name"]'],
|
||||
runtime=SimpleNamespace(state=None),
|
||||
thread_data={"workspace_path": str(tmp_path)},
|
||||
size_prober=lambda *args: 8,
|
||||
)
|
||||
leaf = verdict["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (holds, holds)
|
||||
if not holds:
|
||||
assert leaf["detail"] == "CSV header deadline exceeded"
|
||||
command = sandbox.execute_command.call_args
|
||||
assert command.kwargs["timeout"] == 10
|
||||
assert "/usr/bin/timeout 5" in command.args[0]
|
||||
sandbox.read_file.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("columns", [["z", "id", "a"], ['a"b', "line\nbreak", " id "]], ids=["ordered", "escaped"])
|
||||
def test_missing_columns_detail_names_exact_columns(check_csv, columns):
|
||||
leaf = check_csv(b"id,name\n", columns)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
||||
missing = sorted(set(columns) - {"id", "name"})
|
||||
assert leaf["detail"] == f"CSV header is missing required columns: {json.dumps(missing)}"
|
||||
|
||||
|
||||
def test_missing_columns_detail_remains_bounded(check_csv):
|
||||
leaf = check_csv(b"id,name\n", ["a" * 100, "b" * 100])["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (True, False)
|
||||
assert len(leaf["detail"]) == 160
|
||||
assert leaf["detail"].endswith("...")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("suffix", [b"\n", b"", b"\n" + b"1" * 60_000], ids=["complete", "eof", "large-body"])
|
||||
def test_over_budget_header_detail_is_distinct(check_csv, suffix):
|
||||
content = b"id,name," + b"x" * (50_001 - 8 - (1 if suffix.startswith(b"\n") else 0)) + suffix
|
||||
leaf = check_csv(content)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
||||
assert leaf["detail"] == "CSV header exceeds the 50000-byte read budget"
|
||||
|
||||
|
||||
def test_truncated_header_detail_remains_incomplete(check_csv):
|
||||
leaf = check_csv(b"id,name," + b"x" * 60_000)["leaves"][0]
|
||||
assert (leaf["checked"], leaf["holds"]) == (False, False)
|
||||
assert leaf["detail"] == "CSV header incomplete within the read cap"
|
||||
Loading…
x
Reference in New Issue
Block a user