From e4c893f2fb02e24346b167071292b7f4b4695cd3 Mon Sep 17 00:00:00 2001 From: ZJPex <144683084+ZJPex@users.noreply.github.com> Date: Tue, 13 Oct 2026 19:52:22 +0800 Subject: [PATCH] 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 --- README.md | 20 + backend/AGENTS.md | 4 +- .../harness/deerflow/subagents/AGENTS.md | 2 + .../deerflow/subagents/acceptance_checks.py | 138 ++++++- .../deerflow/tools/builtins/task_tool.py | 4 + backend/tests/test_batch_acceptance.py | 35 ++ backend/tests/test_csv_acceptance.py | 343 ++++++++++++++++++ 7 files changed, 538 insertions(+), 8 deletions(-) create mode 100644 backend/tests/test_csv_acceptance.py diff --git a/README.md b/README.md index c1e1e4506d..e9cbb37f1f 100644 --- a/README.md +++ b/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: exists` and `file_written:`, including on remote sandboxes. It fails `file: 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**. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 4733b9031a..254b6c0654 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -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|`, 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|`, 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. diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index f4bb17b798..bfa46a4718 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -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: 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. diff --git a/backend/packages/harness/deerflow/subagents/acceptance_checks.py b/backend/packages/harness/deerflow/subagents/acceptance_checks.py index 5903db6e56..8abc4107ff 100644 --- a/backend/packages/harness/deerflow/subagents/acceptance_checks.py +++ b/backend/packages/harness/deerflow/subagents/acceptance_checks.py @@ -8,6 +8,9 @@ requirement. Leaf families: +- ``file: csv-columns:["id","name"]`` — UTF-8 CSV header only, optional BOM, + exact required columns. Bounded prefix; no data-row or schema validation. + - ``file: 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.+?)\s+(?Pexists|non-empty)$", re.IGNORECASE) _JSON_FILE_LEAF_RE = re.compile(r"^file:(?P.+?)\s+json-valid$", re.IGNORECASE) +_CSV_FILE_LEAF_RE = re.compile(r"^file:(?P.+?)\s+csv-columns:(?P.+)$", re.IGNORECASE) _FILE_WRITTEN_RE = re.compile(r"^file_written:(?P.+)$", re.IGNORECASE) _TESTS_PASSED_RE = re.compile(r"^tests_passed:(?P.+)$", 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: diff --git a/backend/packages/harness/deerflow/tools/builtins/task_tool.py b/backend/packages/harness/deerflow/tools/builtins/task_tool.py index 8107244c66..a3e38998a6 100644 --- a/backend/packages/harness/deerflow/tools/builtins/task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/task_tool.py @@ -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: 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. diff --git a/backend/tests/test_batch_acceptance.py b/backend/tests/test_batch_acceptance.py index 117195128d..613edb83e8 100644 --- a/backend/tests/test_batch_acceptance.py +++ b/backend/tests/test_batch_acceptance.py @@ -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")) diff --git a/backend/tests/test_csv_acceptance.py b/backend/tests/test_csv_acceptance.py new file mode 100644 index 0000000000..ccfd60689a --- /dev/null +++ b/backend/tests/test_csv_acceptance.py @@ -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"