diff --git a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md index 0d3458f0b..dbb1d605c 100644 --- a/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/migrations/AGENTS.md @@ -19,15 +19,15 @@ The legacy branch handles pre-alembic databases that already have at least one D The empty-DB path keeps using `create_all` because `Base.metadata` is the only authoritative schema source — `create_all` renders both SQLite (JSON, type affinity) and Postgres (JSONB, partial indexes) correctly without anyone having to keep a hand-written baseline in lockstep. `0001_baseline.upgrade()` is therefore almost never executed in practice; it exists as a stamp target + chain root. -**Rolling forward compatibility**: the local chain head is `0020_threads_meta_project_id` -(`0018_oauth_identity_pg_partial` → `0019_projects` → `0020_threads_meta_project_id`). +**Rolling forward compatibility**: the local chain head is `0021_batch_acceptance` +(`0018_oauth_identity_pg_partial` → `0019_projects` → `0020_threads_meta_project_id` → `0021_batch_acceptance`). Bootstrap reads `alembic_version` while holding its backend lock and accepts exactly one row. A locally known revision follows the normal upgrade path. The one unknown revision `0019_thread_incarnations` is conditionally allowlisted: bootstrap first requires every current ORM table and column, then logs a warning and leaves the schema untouched. The original rollout shape (0018 plus the two incarnation columns) is now rejected: it lacks `projects` and -`threads_meta.project_id`. Seeding current head is only a positive compatibility +`threads_meta.project_id`, and the batch acceptance columns. Seeding current head is only a positive compatibility fixture; tests must also construct the original 0018-based schema and assert rejection on both the direct startup and SQLite race-recovery paths. The check uses `conn.run_sync` reflection and derives its local floor from `Base.metadata` @@ -137,7 +137,9 @@ on installs that never enabled it. The convention is: - `migrations/versions/0017_personal_access_tokens.py` — creates the personal access token table for programmatic API access - `migrations/versions/0018_oauth_identity_pg_partial.py` — converts `idx_users_oauth_identity` to a partial index on Postgres (`postgresql_where`), matching what `UserRow.__table_args__` already builds via `create_all`; `0001_baseline` never applied the predicate on Postgres, so every `alembic upgrade head`-provisioned deployment carried a full index until this revision. Postgres-only, idempotent (checks `pg_index.indpred` directly), no-op on SQLite (already partial via `sqlite_where`) and on a DB where the index doesn't exist yet. Originally generated as 0017 and renumbered to 0018 after 0017_personal_access_tokens merged first and kept that slot - `migrations/versions/0019_projects.py` — creates the `projects` table (id/user_id/name/instructions/presentation/status + timestamps) for the Projects Phase-1 organization feature; chains after `0018_oauth_identity_pg_partial` -- `migrations/versions/0020_threads_meta_project_id.py` — adds nullable `threads_meta.project_id` plus `ix_threads_meta_project_id` (no FK by design: project delete clears membership first, and the reserved `deerflow_project_id` metadata key stays in sync); chains after `0019_projects` and is the current head. The `0019_` numeric prefix is reused by the reserved out-of-tree `0019_thread_incarnations` — see the rolling-forward section above +- `migrations/versions/0020_threads_meta_project_id.py` — adds nullable `threads_meta.project_id` plus `ix_threads_meta_project_id` (no FK by design: project delete clears membership first, and the reserved `deerflow_project_id` metadata key stays in sync); chains after `0019_projects`. The `0019_` numeric prefix is reused by the reserved out-of-tree `0019_thread_incarnations` — see the rolling-forward section above - `persistence/bootstrap.py` — `bootstrap_schema(engine, backend=...)`, the three-branch provisioning decision, locked revision validation, and the narrow 0019 forward-compatibility exception - `extensions/loader.py::load_extensions` — registers each spec's `table_prefix` with `register_extension_table_prefix()` - Tests: `tests/test_persistence_bootstrap.py` (branches), `tests/test_persistence_bootstrap_concurrency.py` (concurrency), `tests/test_persistence_bootstrap_regression.py` (issue #3682), `tests/test_persistence_migrations_env.py` (filter, including extension-owned tables), `tests/test_extension_loader.py::TestTablePrefixRegistration` (spec-to-filter wiring), `tests/blocking_io/test_persistence_bootstrap.py` (asyncio.to_thread anchor), `tests/test_migration_0004_run_ownership_dedupe.py` + `tests/test_migration_0007_scheduled_run_active_dedupe.py` (dedupe-before-unique-index pre-steps) + +- `migrations/versions/0021_batch_acceptance.py` — adds nullable per-item acceptance criteria and verdict JSON columns after `0020_threads_meta_project_id`; legacy rows remain unchecked. diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0021_batch_acceptance.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0021_batch_acceptance.py new file mode 100644 index 000000000..ddf417d00 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0021_batch_acceptance.py @@ -0,0 +1,28 @@ +"""Persist optional durable batch acceptance criteria and verdicts. + +Revision ID: 0021_batch_acceptance +Revises: 0020_threads_meta_project_id +""" + +from __future__ import annotations + +import sqlalchemy as sa + +revision = "0021_batch_acceptance" +down_revision = "0020_threads_meta_project_id" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + from deerflow.persistence.migrations._helpers import safe_add_column + + safe_add_column("subagent_batch_items", sa.Column("acceptance_criteria", sa.JSON(), nullable=True)) + safe_add_column("subagent_batch_items", sa.Column("acceptance_verdict", sa.JSON(), nullable=True)) + + +def downgrade() -> None: + from deerflow.persistence.migrations._helpers import safe_drop_column + + safe_drop_column("subagent_batch_items", "acceptance_verdict") + safe_drop_column("subagent_batch_items", "acceptance_criteria") diff --git a/backend/packages/harness/deerflow/persistence/subagent_batches/model.py b/backend/packages/harness/deerflow/persistence/subagent_batches/model.py index 4c4b9da5e..7208941d6 100644 --- a/backend/packages/harness/deerflow/persistence/subagent_batches/model.py +++ b/backend/packages/harness/deerflow/persistence/subagent_batches/model.py @@ -47,6 +47,8 @@ class SubagentBatchItemRow(Base): item_key: Mapped[str] = mapped_column(String(128)) position: Mapped[int] = mapped_column(Integer) prompt: Mapped[str] = mapped_column(Text) + acceptance_criteria: Mapped[list[str] | None] = mapped_column(JSON, nullable=True) + acceptance_verdict: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) status: Mapped[str] = mapped_column(String(24), index=True) attempt: Mapped[int] = mapped_column(Integer, default=0) lease_owner: Mapped[str | None] = mapped_column(String(128), nullable=True) diff --git a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py index 9775c469d..238b22d60 100644 --- a/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py +++ b/backend/packages/harness/deerflow/persistence/subagent_batches/sql.py @@ -10,6 +10,9 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from deerflow.persistence.subagent_batches.model import SubagentBatchItemRow, SubagentBatchRow +from deerflow.subagents.acceptance_checks import AcceptanceVerdict, validate_acceptance_verdict +from deerflow.subagents.batch_runtime import BatchItemInput +from deerflow.subagents.report_contract import normalize_acceptance_criteria from deerflow.utils.time import coerce_iso BATCH_ACTIVE_STATUSES = ("queued", "running", "paused") @@ -44,6 +47,7 @@ _ITEM_PUBLIC_FIELDS = ( "error", "stop_reason", "token_usage", + "acceptance_criteria", "started_at", "completed_at", "created_at", @@ -81,6 +85,7 @@ class SubagentBatchRepository: @staticmethod def _item_dict(row: SubagentBatchItemRow, *, include_result: bool = False) -> dict[str, Any]: data = {key: getattr(row, key) for key in _ITEM_PUBLIC_FIELDS} + data["acceptance_verdict"] = validate_acceptance_verdict(row.acceptance_verdict) if include_result: data["result"] = row.result for key in _ITEM_TIMESTAMP_FIELDS: @@ -99,7 +104,7 @@ class SubagentBatchRepository: submission_key: str, title: str, subagent_type: str, - items: list[dict[str, str]], + items: list[BatchItemInput], max_live_items: int, max_running_items: int, max_attempts: int, @@ -131,6 +136,7 @@ class SubagentBatchRepository: item_key=item["key"], position=position, prompt=item["prompt"], + acceptance_criteria=normalize_acceptance_criteria(item.get("acceptance_criteria")) or None, status="pending", attempt=0, result_truncated=False, @@ -399,6 +405,7 @@ class SubagentBatchRepository: token_usage: dict[str, Any] | None, model_name: str | None, completed_at: datetime, + acceptance_verdict: AcceptanceVerdict | None = None, ) -> bool: async with self._sf() as session: item = ( @@ -422,12 +429,14 @@ class SubagentBatchRepository: item.stop_reason = stop_reason item.token_usage = token_usage item.updated_at = completed_at + item.acceptance_verdict = None if cancelled: item.status = "cancelled" item.error = "Cancelled by user" item.completed_at = completed_at elif succeeded: item.status = "succeeded" + item.acceptance_verdict = validate_acceptance_verdict(acceptance_verdict) item.result = result item.result_preview = result_preview item.result_truncated = result_truncated @@ -566,6 +575,7 @@ class SubagentBatchRepository: item.result = None item.result_preview = None item.result_truncated = False + item.acceptance_verdict = None item.completed_at = None item.cancel_requested_at = None item.updated_at = now diff --git a/backend/packages/harness/deerflow/subagents/AGENTS.md b/backend/packages/harness/deerflow/subagents/AGENTS.md index 24f88e34e..3f8acffee 100644 --- a/backend/packages/harness/deerflow/subagents/AGENTS.md +++ b/backend/packages/harness/deerflow/subagents/AGENTS.md @@ -1,5 +1,17 @@ ### Subagent System (`packages/harness/deerflow/subagents/`) +**Durable batch acceptance**: `batch_task` normalizes optional per-item criteria +before persistence (empty becomes null; 20 items × 500 neutralized characters), +sharing `normalize_acceptance_criteria` with the executor and checker. +Completed items reuse `acceptance_checks` through `batch_acceptance.py`, with +owner-scoped thread paths, sandbox authorization and a client lease. Admission +and checks share `parse_file_criterion` on the effective normalized list. Blocking +reads drain before release on cancellation, and the batch item lease is renewed +while checking. The nullable validated verdict survives repository queries and +JSONL exports independently of execution status. No criteria, checker errors, +and legacy rows have no verdict; they are not accepted by implication. Failed +executions are not checked, and acceptance never changes automatic retry policy. + **Built-in Agents**: `general-purpose` (all tools except `task`) and `bash` (command specialist) **Registry and managed definitions**: Runtime resolution is built-in → `config.yaml custom_agents` → enabled administrator-managed definitions, followed by explicit `subagents.agents.` overrides. Managed definitions are deployment-wide, persist through the same `agent_storage.backend` selection as Custom Agent definitions, and remain stored but are excluded from runtime when a built-in or later-added config definition owns the same name. The default Lead Agent sees the whole enabled catalog. A Custom Agent's `allowed_subagents` is snapshotted into run metadata (`None` = all, `[]` = hard deny, list = allowlist) and must filter both prompt discovery and `task` execution; never reload caller policy from mutable agent config inside the tool. **Benefit-based routing policy**: Enabling subagents exposes delegation as an optimization, not a default response to complexity. The lead prompt defaults to direct execution and permits `task` only when parallel latency, specialist capability, or context-isolation benefit clearly exceeds startup, duplicate-discovery, synthesis, state-conflict, and side-effect costs. Inter-agent output dependencies and overlapping mutable state are hard vetoes for parallel dispatch, while duplicate discovery and a cheap direct path remain costs rather than categorical vetoes; a bounded sequential chain may run in one subagent when specialist or context-isolation benefit clearly wins. Parallel scopes must be independent and non-overlapping, the lead uses the fewest useful subagents, and every later batch is re-evaluated while retaining any within-batch parallel benefit. When the enforced per-response limit is 1, the rendered prompt removes parallel and multi-batch benefit guidance and permits delegation only for material specialist or context-isolation benefit. Keep this policy aligned across `lead_agent/prompt.py`, the `task` tool description, and both built-in role descriptions; routing regressions are pinned in `tests/test_subagent_routing_prompt.py`, `tests/test_subagent_prompt_security.py`, and `tests/test_lead_agent_prompt.py`. diff --git a/backend/packages/harness/deerflow/subagents/acceptance_checks.py b/backend/packages/harness/deerflow/subagents/acceptance_checks.py index a56a66163..65989ffe3 100644 --- a/backend/packages/harness/deerflow/subagents/acceptance_checks.py +++ b/backend/packages/harness/deerflow/subagents/acceptance_checks.py @@ -63,7 +63,7 @@ from collections.abc import Callable, Mapping from typing import Any, TypedDict from deerflow.config.paths import VIRTUAL_PATH_PREFIX -from deerflow.subagents.report_contract import MAX_ACCEPTANCE_CRITERIA, MAX_CRITERION_CHARS +from deerflow.subagents.report_contract import MAX_ACCEPTANCE_CRITERIA, normalize_acceptance_criteria CHECK_SOURCE = "acceptance_checklist" CHECK_REQUIREMENT = "delegation_acceptance_criteria" @@ -139,6 +139,22 @@ class AcceptanceVerdict(TypedDict): all_hold: bool # every leaf checked and holds +def parse_file_criterion(criterion: str) -> tuple[str, str] | None: + """Classify normalized file criteria for both sandbox admission and checks. + + Keep admission on the same Unicode-aware patterns as the checker so no + recognized spelling can fall through to an unowned lazy acquisition. + """ + file_match = _FILE_LEAF_RE.match(criterion) + if file_match is not None: + family = "file_exists" if file_match.group("mode").lower() == "exists" else "file_non_empty" + return family, file_match.group("path") + written_match = _FILE_WRITTEN_RE.match(criterion) + if written_match is not None: + return "file_written", written_match.group("path") + return None + + def _bound_detail(text: str) -> str: cleaned = " ".join(text.split()) if len(cleaned) <= _DETAIL_MAX_CHARS: @@ -1254,29 +1270,10 @@ def check_acceptance_criteria( Returns ``None`` when no usable criterion exists (caller stamps nothing). Synchronous: the async call site offloads via ``asyncio.to_thread`` — - ``content_reader`` performs sandbox IO. Criteria hygiene mirrors - ``report_contract.render_acceptance_criteria_block`` (strip, drop empties, - cap count/length) so the checked list matches the delegated list. + ``content_reader`` performs sandbox IO. The shared normalizer keeps the + checked list identical to the persisted and delegated list. """ - if not acceptance_criteria: - return None - # Lazy import: the sanitizer lives in agents.middlewares, and this package - # is imported in cycles with deerflow.agents (same pattern as - # report_contract). Criterion text is model-supplied untrusted data; it - # must be neutralized here exactly as render_acceptance_criteria_block - # does, or a blocked tag in a criterion would be reintroduced into the - # lead-visible result text by render_acceptance_section. - from deerflow.agents.middlewares.input_sanitization_middleware import neutralize_untrusted_tags - - criteria: list[str] = [] - for criterion in acceptance_criteria: - if not isinstance(criterion, str): - continue - cleaned = criterion.strip()[:MAX_CRITERION_CHARS].strip() - if cleaned: - criteria.append(neutralize_untrusted_tags(cleaned)) - if len(criteria) >= MAX_ACCEPTANCE_CRITERIA: - break + criteria = normalize_acceptance_criteria(acceptance_criteria) if not criteria: return None @@ -1292,15 +1289,11 @@ def check_acceptance_criteria( readable_prober = _probe_file_readable leaves: list[AcceptanceLeaf] = [] for criterion in criteria: - file_match = _FILE_LEAF_RE.match(criterion) - written_match = _FILE_WRITTEN_RE.match(criterion) + file_criterion = parse_file_criterion(criterion) tests_match = _TESTS_PASSED_RE.match(criterion) - if file_match is not None: - mode = file_match.group("mode").lower() - family = "file_exists" if mode == "exists" else "file_non_empty" - leaf = _check_file_leaf(family, file_match.group("path"), runtime=runtime, thread_data=thread_data, content_reader=content_reader, size_prober=size_prober, readable_prober=readable_prober) - elif written_match is not None: - leaf = _check_file_leaf("file_written", written_match.group("path"), runtime=runtime, thread_data=thread_data, content_reader=content_reader, size_prober=size_prober, readable_prober=readable_prober) + 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) 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/subagents/batch_acceptance.py b/backend/packages/harness/deerflow/subagents/batch_acceptance.py new file mode 100644 index 000000000..a401d16f0 --- /dev/null +++ b/backend/packages/harness/deerflow/subagents/batch_acceptance.py @@ -0,0 +1,58 @@ +"""Run the existing checklist for a durable item without a parent tool runtime.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +from deerflow.config.app_config import AppConfig +from deerflow.subagents.acceptance_checks import AcceptanceVerdict, check_acceptance_criteria, parse_file_criterion +from deerflow.subagents.report_contract import normalize_acceptance_criteria + + +def _thread_data(thread_id: str, user_id: str) -> dict[str, str]: + from deerflow.config.paths import get_paths + + paths = get_paths() + return { + "workspace_path": str(paths.sandbox_work_dir(thread_id, user_id=user_id)), + "uploads_path": str(paths.sandbox_uploads_dir(thread_id, user_id=user_id)), + "outputs_path": str(paths.sandbox_outputs_dir(thread_id, user_id=user_id)), + } + + +async def check_batch_acceptance( + criteria: list[str], + *, + batch: dict[str, Any], + app_config: AppConfig, + bash_executions: list[dict[str, Any]] | None, +) -> AcceptanceVerdict | None: + from deerflow.authz.sandbox_authz import authorize_sandbox_execution_async + from deerflow.sandbox.lease import SANDBOX_LEASE_OWNER_CONTEXT_KEY, acquire_sandbox_client_lease, run_sync_lifecycle_operation + from deerflow.sandbox.sandbox_provider import get_sandbox_provider + + criteria = await run_sync_lifecycle_operation(normalize_acceptance_criteria, criteria) + if not criteria: + return None + thread_id, user_id = batch["thread_id"], batch["user_id"] + spec = batch["execution_spec"] + context = {key: spec.get(key) for key in ("user_role", "oauth_provider", "oauth_id", "channel_user_id", "is_internal", "authz_attributes")} + context.update(thread_id=thread_id, user_id=user_id) + thread_data = await run_sync_lifecycle_operation(_thread_data, thread_id, user_id) + runtime = SimpleNamespace(state={"thread_data": thread_data}, context=context, config={"configurable": {"thread_id": thread_id}}) + lease = None + try: + # Evidence-only / unsupported conditions need no sandbox. File checks + # use an authorized, owner-scoped holder of the shared thread sandbox. + if any(parse_file_criterion(criterion) is not None for criterion in criteria): + await authorize_sandbox_execution_async(context=context, app_config=app_config) + provider = await run_sync_lifecycle_operation(get_sandbox_provider) + lease = await acquire_sandbox_client_lease(provider, thread_id, user_id=user_id, owner_prefix="batch-acceptance") + runtime.state["sandbox"] = {"sandbox_id": lease.sandbox_id} + context[SANDBOX_LEASE_OWNER_CONTEXT_KEY] = lease.owner_id + # Drain blocking reads before releasing the holder, including shutdown. + return await run_sync_lifecycle_operation(check_acceptance_criteria, criteria, runtime=runtime, thread_data=thread_data, bash_executions=bash_executions) + finally: + if lease is not None: + await lease.release() diff --git a/backend/packages/harness/deerflow/subagents/batch_runtime.py b/backend/packages/harness/deerflow/subagents/batch_runtime.py index 7ce53522d..69dc7bf0e 100644 --- a/backend/packages/harness/deerflow/subagents/batch_runtime.py +++ b/backend/packages/harness/deerflow/subagents/batch_runtime.py @@ -4,7 +4,13 @@ from __future__ import annotations import threading from dataclasses import dataclass -from typing import Any, Protocol +from typing import Any, NotRequired, Protocol, TypedDict + + +class BatchItemInput(TypedDict): + key: str + prompt: str + acceptance_criteria: NotRequired[list[str] | None] @dataclass(frozen=True) @@ -16,7 +22,7 @@ class BatchSubmitRequest: submission_key: str title: str subagent_type: str - items: list[dict[str, str]] + items: list[BatchItemInput] max_live_items: int | None max_running_items: int | None execution_spec: dict[str, Any] diff --git a/backend/packages/harness/deerflow/subagents/batch_service.py b/backend/packages/harness/deerflow/subagents/batch_service.py index ed1810b04..2abcc2b47 100644 --- a/backend/packages/harness/deerflow/subagents/batch_service.py +++ b/backend/packages/harness/deerflow/subagents/batch_service.py @@ -10,6 +10,7 @@ from typing import Any from deerflow.config.app_config import AppConfig, get_app_config from deerflow.config.subagent_batches_config import SubagentBatchesConfig from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig +from deerflow.subagents.batch_acceptance import check_batch_acceptance from deerflow.subagents.batch_runtime import BatchSubmitRequest from deerflow.subagents.capacity import SubagentExecutionCapacity from deerflow.subagents.config import SubagentConfig, resolve_subagent_model_name @@ -216,6 +217,7 @@ class SubagentBatchService: is_internal=spec.get("is_internal") is True, authz_attributes=spec.get("authz_attributes"), execution_capacity=self._execution_capacity, + acceptance_criteria=item.get("acceptance_criteria"), ) prompt = f"Durable batch item key: {item['item_key']}\nThis item may be retried after a worker crash. Keep side effects idempotent and use the item key as the idempotency identity.\n\n{item['prompt']}" execution_id = executor.execute_async(prompt, task_id=item_id) @@ -276,6 +278,16 @@ class SubagentBatchService: truncated = len(raw_result) > self._config.max_result_chars stored_result = raw_result[: self._config.max_result_chars] if raw_result else None preview = raw_result[: self._config.result_preview_max_chars] if raw_result else None + acceptance_verdict = None + if result.status is SubagentStatus.COMPLETED and item.get("acceptance_criteria"): + try: + valid, acceptance_verdict = await self._check_acceptance_with_lease(item, result, app_config) + if not valid: + return + except Exception: + # Advisory like ordinary task acceptance: an unavailable + # checker must not discard useful work or trigger a retry. + logger.warning("Batch acceptance check failed; result remains unchecked (item_id=%s)", item_id, exc_info=True) await self._repository.finalize_item( item_id, lease_owner=self._lease_owner, @@ -288,6 +300,7 @@ class SubagentBatchService: token_usage=_usage(result.token_usage_records), model_name=effective_model, completed_at=datetime.now(UTC), + acceptance_verdict=acceptance_verdict, ) except asyncio.CancelledError: if execution_id is not None: @@ -318,3 +331,39 @@ class SubagentBatchService: self._item_batches.pop(item_id, None) if execution_id is not None: cleanup_background_task(execution_id) + + async def _check_acceptance_with_lease(self, item, result, app_config): + """Keep a completed execution leased until its advisory check drains.""" + + async def renew(): + lease = await self._repository.renew_item_lease( + item["id"], + lease_owner=self._lease_owner, + lease_seconds=self._config.lease_seconds, + now=datetime.now(UTC), + ) + return lease["valid"] + + if not await renew(): + return False, None + check = asyncio.create_task( + check_batch_acceptance( + item["acceptance_criteria"], + batch=item["batch"], + app_config=app_config, + bash_executions=getattr(result, "bash_executions", None), + ) + ) + try: + while True: + done, _ = await asyncio.wait({check}, timeout=max(1.0, self._config.lease_seconds / 3)) + if done: + return True, check.result() + if not await renew(): + return False, None + finally: + if not check.done(): + check.cancel() + # The checklist's sandbox offload drains before releasing its + # holder, even when shutdown or a lost lease cancels this task. + await asyncio.gather(check, return_exceptions=True) diff --git a/backend/packages/harness/deerflow/subagents/report_contract.py b/backend/packages/harness/deerflow/subagents/report_contract.py index b7e096075..f964c1590 100644 --- a/backend/packages/harness/deerflow/subagents/report_contract.py +++ b/backend/packages/harness/deerflow/subagents/report_contract.py @@ -113,21 +113,14 @@ def build_acceptance_criteria_system_note(*, receipts_enabled: bool = True) -> s ) -def render_acceptance_criteria_block(acceptance_criteria: list[str] | None) -> str: - """Render lead-supplied acceptance criteria as data for the task message. +def normalize_acceptance_criteria(acceptance_criteria: list[str] | None) -> list[str]: + """Return the bounded, neutralized list shared by storage and execution. - Returns "" when there is nothing usable. Entries are stripped, empties - dropped, the list/item sizes capped, and each entry neutralized via - :func:`neutralize_untrusted_tags` before interpolation, so the stored - state itself carries no live framework/injection tags. The block uses a - plain-text header rather than an ```` tag on purpose: - the task ``HumanMessage`` is sanitized by ``InputSanitizationMiddleware`` - at model-call time, which HTML-escapes denylisted framework tags — a tag - here would reach the model only in escaped form, while plain markdown - survives intact. + Reapplying this operation to persisted criteria preserves their text: + escaping can expand tags, so bound the neutralized output as well. """ if not acceptance_criteria: - return "" + return [] # Lazy import: the executor package is imported in cycles with # ``deerflow.agents``; resolving the sanitizer at call time keeps module # init order-independent (same pattern as build_report_contract_section). @@ -139,9 +132,24 @@ def render_acceptance_criteria_block(acceptance_criteria: list[str] | None) -> s continue cleaned = criterion.strip()[:MAX_CRITERION_CHARS].strip() if cleaned: - criteria.append(neutralize_untrusted_tags(cleaned)) + # Expansion can move the cap into an otherwise allowed tag name + # (e.g. -> = MAX_ACCEPTANCE_CRITERIA: break + return criteria + + +def render_acceptance_criteria_block(acceptance_criteria: list[str] | None) -> str: + """Render normalized criteria as untrusted data for the task message. + + Returns "" when there is nothing usable. The block uses a plain-text + header rather than a framework tag: InputSanitizationMiddleware escapes + those tags when it frames the task HumanMessage as untrusted input. + """ + criteria = normalize_acceptance_criteria(acceptance_criteria) if not criteria: return "" items = "\n".join(f"- {criterion}" for criterion in criteria) diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md index 9c5231a58..1518036f4 100644 --- a/backend/packages/harness/deerflow/tools/AGENTS.md +++ b/backend/packages/harness/deerflow/tools/AGENTS.md @@ -12,7 +12,7 @@ 4. **Subagent tool** (if enabled): - `task` - Delegate to subagent (`prompt`, `subagent_type`, optional `acceptance_criteria`, and an optional model-visible `description` used only as a short progress label). Execution never depends on `description`; lifecycle display falls back to `prompt` when a provider omits it. Subagent reports are self-reports: the docstring directs the lead to expect `[rN]` receipt citations and verifiable handles while `verification.receipts_enabled` (and explicitly qualifies that disabled receipts mean no citations and no citation verdict), to read the delegation ledger's citation cross-check as execution evidence only, and to attach `acceptance_criteria` for objectively checkable outcomes (canonical forms `file: exists|non-empty`, `file_written:`, `tests_passed:`); criteria are handed to the executor and appended to the subagent's task message as untrusted data (see `subagents/report_contract.py`). Polling safety timeouts carry the latest published tool receipts into the terminal task metadata before requesting background cancellation. - - `batch_task`, `batch_status`, `cancel_batch` - Explicit durable batch submission/progress/cancellation. Added only while the startup SQL-backed batch submitter is installed; large results stay in the owner-scoped API/JSONL export rather than the lead context. + - `batch_task`, `batch_status`, `cancel_batch` - Explicit durable batch submission/progress/cancellation. Added only while the startup SQL-backed batch submitter is installed; large results stay in the owner-scoped API/JSONL export rather than the lead context. Items accept optional `acceptance_criteria`; item queries and exports expose the separate `acceptance_verdict`. Progress counts describe execution, not acceptance; unmet and UNVERIFIED conditions never trigger automatic retries. - Direct `create_deerflow_agent` integrations receive cloned tools bound to their explicit `SubagentRuntime`. The bound `task` forwards that runtime's exact execution controller and optional caller-owned `AppConfig` into registry/model/tool resolution and `SubagentExecutor`; bound batch tools use the same config snapshot and resolve only that runtime's submitter before falling back to no other application's active worker. Keep the original tool name/schema unchanged so model contracts and user-tool deduplication remain stable. The ordinary `task` boundary carries one narrow parent-loop middleware recorder into the isolated subagent runtime under separate loop-detection and tool-promotion keys. It schedules only `record_middleware` calls back onto the loop that owns `RunJournal`, keeps an execution-local atomic promotion claim so parallel searches do not double-report one new schema, is fenced and drained once before `task` returns, and never exposes the journal or event store to the child loop. Durable batch tasks have no parent run journal and do not use this bridge. diff --git a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py b/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py index 31e255a9d..6ba9288ba 100644 --- a/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py +++ b/backend/packages/harness/deerflow/tools/builtins/batch_task_tool.py @@ -16,6 +16,7 @@ from pydantic import BaseModel, Field from deerflow.authz.principal import normalize_authz_attributes from deerflow.runtime.user_context import resolve_runtime_user_id from deerflow.subagents.batch_runtime import ( + BatchItemInput, BatchSubmitRequest, SubagentBatchSubmitter, get_subagent_batch_submitter, @@ -27,6 +28,7 @@ from deerflow.tools.types import Runtime class BatchTaskItem(BaseModel): key: str = Field(min_length=1, max_length=128) prompt: str = Field(min_length=1, max_length=100_000) + acceptance_criteria: list[str] | None = Field(default=None, description="Optional completion requirements checked separately from execution status, using the same bounded checklist as task.") _NO_EXPLICIT_BATCH_SUBMITTER = object() @@ -147,9 +149,17 @@ async def batch_task( identifier immediately; it never inserts thousands of results into the lead agent context. Use ``batch_status`` for a compact progress snapshot. + Each item may carry ``acceptance_criteria`` using the same deterministic + checks as ``task``: ``file: exists|non-empty``, ``file_written:``, + or ``tests_passed:`` against recorded execution evidence. + Other conditions are UNVERIFIED. Item queries and JSONL exports include the + separate acceptance verdict; ``succeeded`` only means execution completed. + Retain useful results, repair unmet conditions, and verify consequential + unknowns or preserve uncertainty. Acceptance never triggers automatic retries. + Args: title: Short batch name shown to the user. - items: Stable item keys and self-contained prompts. + items: Stable item keys, self-contained prompts, and optional per-item acceptance_criteria. subagent_type: Native subagent definition used for every item. max_live_items: Optional queued-plus-running item window. max_running_items: Optional per-batch real execution concurrency. @@ -212,7 +222,7 @@ async def batch_task( submission_key=submission_key, title=title.strip()[:256] or "Subagent batch", subagent_type=subagent_type, - items=[item.model_dump() for item in items], + items=[cast(BatchItemInput, item.model_dump(exclude_none=True)) for item in items], max_live_items=max_live_items, max_running_items=max_running_items, execution_spec=execution_spec, @@ -231,6 +241,9 @@ async def batch_task( async def batch_status(runtime: Runtime, batch_id: str) -> str: """Return a compact durable batch progress snapshot. + Counts describe execution status, not acceptance. Inspect item queries or + JSONL exports for the recorded acceptance criteria and verdicts. + Args: batch_id: Server batch identifier returned by ``batch_task``. """ diff --git a/backend/tests/blocking_io/test_batch_acceptance.py b/backend/tests/blocking_io/test_batch_acceptance.py new file mode 100644 index 000000000..4f2ff4657 --- /dev/null +++ b/backend/tests/blocking_io/test_batch_acceptance.py @@ -0,0 +1,83 @@ +"""Batch checklist IO runs off-loop and drains before its sandbox holder closes.""" + +import asyncio +import threading +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from deerflow.config.paths import Paths +from deerflow.subagents import batch_acceptance +from deerflow.subagents.acceptance_checks import check_acceptance_criteria + +pytestmark = pytest.mark.asyncio + + +async def _setup(monkeypatch, tmp_path): + paths = await asyncio.to_thread(Paths, str(tmp_path)) + monkeypatch.setattr("deerflow.config.paths._paths", paths) + probe = tmp_path / "probe.txt" + probe.write_text("actual output") + lease = SimpleNamespace(sandbox_id="local", owner_id="check-lease", release=AsyncMock()) + monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: object()) + monkeypatch.setattr("deerflow.sandbox.lease.acquire_sandbox_client_lease", AsyncMock(return_value=lease)) + return probe, lease + + +def _check(reader): + def check(criteria, **kwargs): + return check_acceptance_criteria(criteria, **kwargs, size_prober=lambda *args: 10, content_reader=reader) + + return check + + +def _kwargs(): + return dict(batch={"thread_id": "t", "user_id": "u", "execution_spec": {}}, app_config=SimpleNamespace(), bash_executions=None) + + +async def test_real_blocking_file_read_is_offloaded(monkeypatch, tmp_path): + probe, lease = await _setup(monkeypatch, tmp_path) + monkeypatch.setattr(batch_acceptance, "check_acceptance_criteria", _check(lambda *args: probe.read_text())) + verdict = await batch_acceptance.check_batch_acceptance(["file:../outputs/report.md exists"], **_kwargs()) + assert verdict["leaves"][0]["checked"] is True + assert verdict["leaves"][0]["holds"] is True + lease.release.assert_awaited_once() + + +async def test_same_blocking_reader_trips_the_gate_on_loop(monkeypatch, tmp_path): + from blockbuster import BlockingError + + probe, _ = await _setup(monkeypatch, tmp_path) + with pytest.raises(BlockingError): + check_acceptance_criteria( + ["file:../outputs/report.md exists"], thread_data={"workspace_path": str(tmp_path / "workspace"), "outputs_path": str(tmp_path / "outputs")}, size_prober=lambda *args: 10, content_reader=lambda *args: probe.read_text() + ) + + +async def test_repeated_cancellation_drains_read_before_releasing_lease(monkeypatch, tmp_path): + probe, lease = await _setup(monkeypatch, tmp_path) + started = asyncio.Event() + unblock = threading.Event() + loop = asyncio.get_running_loop() + + def reader(*args): + loop.call_soon_threadsafe(started.set) + assert unblock.wait(timeout=5) + return probe.read_text() + + monkeypatch.setattr(batch_acceptance, "check_acceptance_criteria", _check(reader)) + task = asyncio.create_task(batch_acceptance.check_batch_acceptance(["file:../outputs/report.md exists"], **_kwargs())) + try: + await asyncio.wait_for(started.wait(), timeout=5) + task.cancel() + await asyncio.sleep(0) + task.cancel() + await asyncio.sleep(0) + assert not task.done() + lease.release.assert_not_awaited() + finally: + unblock.set() + with pytest.raises(asyncio.CancelledError): + await task + lease.release.assert_awaited_once() diff --git a/backend/tests/test_batch_acceptance.py b/backend/tests/test_batch_acceptance.py new file mode 100644 index 000000000..b1105405c --- /dev/null +++ b/backend/tests/test_batch_acceptance.py @@ -0,0 +1,378 @@ +"""Batch submission, real file checks, durable results, and owner-scoped export.""" + +import asyncio +import importlib +import json +from datetime import UTC, datetime, timedelta +from enum import Enum +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +import pytest_asyncio + +from deerflow.config.database_config import DatabaseConfig +from deerflow.config.paths import Paths +from deerflow.config.subagent_batches_config import SubagentBatchesConfig +from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.subagent_batches import SubagentBatchRepository +from deerflow.sandbox.local.local_sandbox_provider import LocalSandboxProvider +from deerflow.subagents import batch_service +from deerflow.subagents.batch_runtime import BatchSubmitRequest +from deerflow.subagents.config import SubagentConfig +from deerflow.tools.builtins.batch_task_tool import BatchTaskItem, bind_batch_tools + + +class SubagentStatus(Enum): + COMPLETED = "completed" + FAILED = "failed" + RUNNING = "running" + + @property + def is_terminal(self): + return self is not SubagentStatus.RUNNING + + +@pytest_asyncio.fixture +async def env(monkeypatch, tmp_path): + paths = Paths(str(tmp_path / "data")) + monkeypatch.setattr("deerflow.config.paths._paths", paths) + paths.ensure_thread_dirs("thread-1", user_id="user-1") + provider = LocalSandboxProvider() + monkeypatch.setattr("deerflow.sandbox.sandbox_provider.get_sandbox_provider", lambda: provider) + monkeypatch.setattr("deerflow.sandbox.tools.get_sandbox_provider", lambda: provider) + monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: []) + monkeypatch.setattr(batch_service, "resolve_subagent_model_name", lambda *args, **kwargs: "model-a") + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path / "db"))) + repo = SubagentBatchRepository(get_session_factory()) + state = SimpleNamespace( + paths=paths, + repo=repo, + provider=provider, + calls=[], + result=SimpleNamespace(status=SubagentStatus.COMPLETED, result="Report claims everything is done", error=None, stop_reason=None, token_usage_records=[], bash_executions=[]), + ) + + class Executor: + def __init__(self, **kwargs): + state.calls.append(kwargs) + + def execute_async(self, prompt, task_id=None): + state.prompt = prompt + return task_id + + monkeypatch.setattr(batch_service, "SubagentExecutor", Executor) + monkeypatch.setattr(batch_service, "SubagentStatus", SubagentStatus) + monkeypatch.setattr(batch_service, "get_background_task_result", lambda _: state.result) + monkeypatch.setattr(batch_service, "cleanup_background_task", lambda _: None) + state.service = batch_service.SubagentBatchService(repository=repo, config=SubagentBatchesConfig(), runtime_config=SubagentRuntimeConfig(), app_config=SimpleNamespace()) + try: + yield state + finally: + await state.service.stop() + await close_engine() + + +async def _submit(env, criteria=None): + item = {"key": "one", "prompt": "Prepare the report"} + if criteria is not None: + item["acceptance_criteria"] = criteria + return await env.service.submit( + BatchSubmitRequest( + user_id="user-1", + thread_id="thread-1", + run_id="run-1", + tool_call_id="call-1", + submission_key="run-1:call-1", + title="Reports", + subagent_type="general-purpose", + items=[item], + max_live_items=1, + max_running_items=1, + execution_spec={"subagent_config": {"name": "general-purpose", "description": "Worker", "system_prompt": "Work carefully"}, "user_role": "member"}, + ) + ) + + +async def _execute(env): + await env.service.run_once(now=datetime.now(UTC)) + await asyncio.gather(*list(env.service._executions.values())) + + +@pytest.mark.asyncio +async def test_tool_preserves_per_item_criteria_and_legacy_shape(env, monkeypatch): + module = importlib.import_module("deerflow.tools.builtins.batch_task_tool") + monkeypatch.setattr(module, "get_available_subagent_names", lambda **kwargs: ["general-purpose"]) + monkeypatch.setattr(module, "get_subagent_config", lambda *args, **kwargs: SubagentConfig(name="general-purpose", description="Worker")) + tools = {tool.name: tool for tool in bind_batch_tools(env.service)} + runtime = SimpleNamespace(state={}, context={"thread_id": "thread-1", "user_id": "user-1"}, config={"metadata": {}}) + command = await tools["batch_task"].coroutine( + runtime=runtime, title="Batch", items=[BatchTaskItem(key="one", prompt="p", acceptance_criteria=["file:../outputs/report.md exists"]), BatchTaskItem(key="two", prompt="q")], subagent_type="general-purpose", tool_call_id="c1" + ) + batch_id = command.update["messages"][0].additional_kwargs["subagent_batch_id"] + items = await env.repo.list_items(batch_id, user_id="user-1") + assert items[0]["acceptance_criteria"] == ["file:../outputs/report.md exists"] + assert items[1]["acceptance_criteria"] is None + + +@pytest.mark.asyncio +async def test_completed_batch_records_mixed_checks_and_exports_after_reopen(env, monkeypatch): + from app.gateway.routers import subagent_batches as router + + output = env.paths.sandbox_outputs_dir("thread-1", user_id="user-1") / "report.md" + output.write_text("Actual report", encoding="utf-8") + criteria = ["file_written:../outputs/report.md", "file:../outputs/missing.csv exists", "claims are correct", "tests_passed:pytest tests/check.py"] + env.result.bash_executions = [{"tool_call_id": "test-run", "tool_name": "bash", "command": "pytest tests/check.py", "output_tail": "3 passed", "status": "success", "shell_persistent": False}] + batch = await _submit(env, criteria) + await _execute(env) + assert env.calls[0]["acceptance_criteria"] == criteria + # Reopen the repository: query and export cannot rely on an in-memory result. + 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 + leaves = item["acceptance_verdict"]["leaves"] + assert [(leaf["checked"], leaf["holds"]) for leaf in leaves] == [(True, True), (True, False), (False, False), (True, True)] + 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) == [] + assert await repo.list_items(batch["id"], user_id="other") is None + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(subagent_batch_repo=repo))) + monkeypatch.setattr(router, "get_current_user", AsyncMock(return_value="user-1")) + listed = await router.list_batch_items.__wrapped__(thread_id="thread-1", batch_id=batch["id"], request=request, offset=0, limit=100, status=None) + assert listed[0]["acceptance_verdict"] == item["acceptance_verdict"] + exported = await router.export_batch_results.__wrapped__(thread_id="thread-1", batch_id=batch["id"], request=request) + rows = [json.loads(line) async for line in exported.body_iterator] + assert rows[0]["acceptance_verdict"] == item["acceptance_verdict"] + assert rows[0]["result"] == env.result.result + assert output.read_text() == "Actual report" + + +@pytest.mark.asyncio +async def test_no_criteria_skips_checker_and_preserves_success(env, monkeypatch): + checker = AsyncMock(side_effect=AssertionError("no checklist")) + monkeypatch.setattr(batch_service, "check_batch_acceptance", checker) + batch = await _submit(env) + await _execute(env) + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "succeeded" + assert item["acceptance_verdict"] is None + checker.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_criteria_survive_lease_recovery_and_submission_replay(env): + criteria = ["file:../outputs/report.md exists"] + first = await _submit(env, criteria) + replay = await _submit(env, ["different"]) + assert replay["id"] == first["id"] + now = datetime.now(UTC) + one = (await env.repo.claim_items(now=now, lease_owner="dead", lease_seconds=1, limit=1))[0] + two = (await env.repo.claim_items(now=now + timedelta(seconds=2), lease_owner="new", lease_seconds=60, limit=1))[0] + assert one["id"] == two["id"] + assert two["acceptance_criteria"] == criteria + + +@pytest.mark.asyncio +async def test_file_check_cannot_confirm_another_users_file(env): + env.paths.ensure_thread_dirs("thread-1", user_id="other") + other_file = env.paths.sandbox_outputs_dir("thread-1", user_id="other") / "secret.txt" + other_file.write_text("private") + batch = await _submit(env, [f"file:{other_file} exists"]) + await _execute(env) + leaf = (await env.repo.list_items(batch["id"], user_id="user-1"))[0]["acceptance_verdict"]["leaves"][0] + assert leaf["checked"] is False + assert leaf["holds"] is False + + +@pytest.mark.asyncio +async def test_checker_error_does_not_retry_successful_execution(env, monkeypatch): + module = importlib.import_module("deerflow.subagents.batch_acceptance") + monkeypatch.setattr(module, "check_acceptance_criteria", lambda *args, **kwargs: (_ for _ in ()).throw(RuntimeError("checker failed"))) + batch = await _submit(env, ["claims are correct"]) + await _execute(env) + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "succeeded" + assert item["acceptance_verdict"] is None + assert item["result_preview"] == env.result.result + + +@pytest.mark.asyncio +async def test_failed_execution_is_not_checked(env, monkeypatch): + checker = AsyncMock(side_effect=AssertionError("failed execution must not be checked")) + monkeypatch.setattr(batch_service, "check_batch_acceptance", checker) + env.result.status = SubagentStatus.FAILED + env.result.error = "worker failed" + batch = await _submit(env, ["file:../outputs/report.md exists"]) + await _execute(env) + checker.assert_not_awaited() + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "queued" + assert item["acceptance_verdict"] is None + + +@pytest.mark.asyncio +async def test_slow_checker_renews_lease_and_stops_after_losing_it(env, monkeypatch): + started = asyncio.Event() + drained = asyncio.Event() + + async def check(*args, **kwargs): + started.set() + try: + await asyncio.Event().wait() + finally: + drained.set() + + monkeypatch.setattr(batch_service, "check_batch_acceptance", check) + env.service._config = env.service._config.model_copy(update={"lease_seconds": 3}) + batch = await _submit(env, ["quality"]) + await env.service.run_once(now=datetime.now(UTC)) + execution = list(env.service._executions.values())[0] + await asyncio.wait_for(started.wait(), timeout=5) + # First renewal admitted the checker. Losing the next renewal must drain + # it and prevent publication, rather than producing a stale verdict. + renew = AsyncMock(return_value={"valid": False, "cancel_requested": True}) + monkeypatch.setattr(env.repo, "renew_item_lease", renew) + await asyncio.wait_for(execution, timeout=5) + renew.assert_awaited_once() + assert drained.is_set() + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "leased" + assert item["acceptance_verdict"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("prefix", ["file", "FILE", "File_Written"]) +async def test_denied_sandbox_keeps_result_unchecked_without_acquiring(env, monkeypatch, prefix): + from deerflow.sandbox.exceptions import SandboxAuthorizationError + + authorize = AsyncMock(side_effect=SandboxAuthorizationError()) + acquire = AsyncMock(side_effect=AssertionError("must not acquire")) + monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution_async", authorize) + monkeypatch.setattr("deerflow.sandbox.lease.acquire_sandbox_client_lease", acquire) + batch = await _submit(env, [f"{prefix}:../outputs/report.md exists"]) + await _execute(env) + acquire.assert_not_awaited() + assert authorize.await_args.kwargs["context"]["user_id"] == "user-1" + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "succeeded" + assert item["acceptance_verdict"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", ["oversized", "escaped", "truncated_tag", "empty"]) +async def test_stored_delegated_checked_and_exported_criteria_agree(env, monkeypatch, case): + from app.gateway.routers import subagent_batches as router + from deerflow.subagents.acceptance_checks import check_acceptance_criteria + from deerflow.subagents.report_contract import render_acceptance_criteria_block + + if case == "oversized": + criteria = ["", " ", None, 42] + [" " + "x" * 1000 + " "] * 25 + expected = ["x" * 500] * 20 + elif case == "escaped": + criteria = ["" * 80] + expected = [("<system>" * 80)[:500]] + elif case == "truncated_tag": + # Escaping the earlier tags shifts the final cap into an allowed + # tag name, exposing a bare blocked prefix (" * 35 + "xxx"] + expected = ["<system>" * 35 + "xxx<sys"] + else: + criteria = ["", " \t ", None, 42] + expected = None + + batch = await _submit(env, criteria) + repo = SubagentBatchRepository(get_session_factory()) + # Assert the write boundary, before either the executor or checker runs. + assert (await repo.list_items(batch["id"], user_id="user-1"))[0]["acceptance_criteria"] == expected + await _execute(env) + assert env.calls[0]["acceptance_criteria"] == expected + item = (await repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "succeeded" + if expected is None: + assert item["acceptance_verdict"] is None + assert render_acceptance_criteria_block(expected) == "" + else: + assert [leaf["criterion"] for leaf in item["acceptance_verdict"]["leaves"]] == expected + assert render_acceptance_criteria_block(expected).split("\n- ")[1:] == expected + assert render_acceptance_criteria_block(criteria).split("\n- ")[1:] == expected + assert [leaf["criterion"] for leaf in check_acceptance_criteria(criteria)["leaves"]] == expected + request = SimpleNamespace(app=SimpleNamespace(state=SimpleNamespace(subagent_batch_repo=repo))) + monkeypatch.setattr(router, "get_current_user", AsyncMock(return_value="user-1")) + response = await router.export_batch_results.__wrapped__(thread_id="thread-1", batch_id=batch["id"], request=request) + exported = [json.loads(line) async for line in response.body_iterator] + assert exported[0]["acceptance_criteria"] == expected + assert exported[0]["acceptance_verdict"] == item["acceptance_verdict"] + + +_FILE_CRITERIA = [ + "file:../outputs/report.md exists", + "FILE:../outputs/report.md non-empty", + "fıle:../outputs/report.md exists", + "FİLE:../outputs/report.md exists", + "file_written:../outputs/report.md", + "fıle_written:../outputs/report.md", + "FİLE_WRİTTEN:../outputs/report.md", +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("criterion", _FILE_CRITERIA) +async def test_caller_sandbox_deny_applies_to_every_file_spelling(env, monkeypatch, criterion): + from deerflow.authz import sandbox_authz + from deerflow.config.authorization_config import AuthorizationConfig, AuthorizationProviderConfig + + env.service._app_config = SimpleNamespace( + authorization=AuthorizationConfig( + enabled=True, + default_role="member", + provider=AuthorizationProviderConfig(use="deerflow.authz.rbac:RbacAuthorizationProvider", config={"roles": {"member": {"sandbox": {"allow": False}}}}), + ) + ) + # Embedded callers can have a different policy from the process global. + monkeypatch.setattr("deerflow.sandbox.tools.safe_app_config", lambda: None) + authorize = AsyncMock(wraps=sandbox_authz.authorize_sandbox_execution_async) + acquire = Mock(wraps=env.provider.acquire) + monkeypatch.setattr(sandbox_authz, "authorize_sandbox_execution_async", authorize) + monkeypatch.setattr(env.provider, "acquire", acquire) + batch = await _submit(env, [criterion]) + await _execute(env) + authorize.assert_awaited_once() + acquire.assert_not_called() + item = (await env.repo.list_items(batch["id"], user_id="user-1"))[0] + assert item["status"] == "succeeded" + assert item["acceptance_verdict"] is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("criterion", _FILE_CRITERIA) +async def test_allowed_file_spellings_read_under_a_released_holder(env, monkeypatch, criterion): + from deerflow.sandbox import lease + from deerflow.subagents.batch_acceptance import check_batch_acceptance + + (env.paths.sandbox_outputs_dir("thread-1", user_id="user-1") / "report.md").write_text("Actual report") + monkeypatch.setattr("deerflow.sandbox.tools.safe_app_config", lambda: None) + acquire = AsyncMock(wraps=lease.acquire_sandbox_client_lease) + release = Mock(wraps=env.provider.release) + monkeypatch.setattr(lease, "acquire_sandbox_client_lease", acquire) + monkeypatch.setattr(env.provider, "release", release) + verdict = await check_batch_acceptance([criterion], batch={"thread_id": "thread-1", "user_id": "user-1", "execution_spec": {}}, app_config=SimpleNamespace(), bash_executions=None) + acquire.assert_awaited_once() + release.assert_called_once() + assert verdict["leaves"][0]["checked"] is True + assert verdict["leaves"][0]["holds"] is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize("criteria", [["quality"] * 20 + ["file:../outputs/report.md exists"], ["file:missing mode"], ["file:" + "x" * 500 + " exists"]]) +async def test_only_effective_file_checks_request_sandbox_access(env, monkeypatch, criteria): + from deerflow.subagents.batch_acceptance import check_batch_acceptance + + authorize = AsyncMock(side_effect=AssertionError("no effective file check")) + acquire = Mock(side_effect=AssertionError("no sandbox acquisition")) + monkeypatch.setattr("deerflow.authz.sandbox_authz.authorize_sandbox_execution_async", authorize) + monkeypatch.setattr(env.provider, "acquire", acquire) + verdict = await check_batch_acceptance(criteria, batch={"thread_id": "thread-1", "user_id": "user-1", "execution_spec": {}}, app_config=SimpleNamespace(), bash_executions=None) + assert all(leaf["family"] == "undecidable" for leaf in verdict["leaves"]) + authorize.assert_not_awaited() + acquire.assert_not_called() diff --git a/backend/tests/test_migration_0004_run_ownership_dedupe.py b/backend/tests/test_migration_0004_run_ownership_dedupe.py index 5f2ef6be3..16c3ee605 100644 --- a/backend/tests/test_migration_0004_run_ownership_dedupe.py +++ b/backend/tests/test_migration_0004_run_ownership_dedupe.py @@ -157,7 +157,7 @@ async def test_migration_dedupes_duplicate_active_rows_before_unique_index(tmp_p with sqlite3.connect(db_path) as raw: version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() # Bootstrap upgrades through the later revisions after 0004. - assert version_row[0] == "0020_threads_meta_project_id" + assert version_row[0] == "0021_batch_acceptance" # Sanity: the invariant the index enforces is now true — at most one # active row per thread. diff --git a/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py b/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py index ebdcd2f42..65c9527c9 100644 --- a/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py +++ b/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py @@ -173,7 +173,7 @@ async def test_migration_supersedes_duplicate_active_runs_before_unique_index(tm with sqlite3.connect(db_path) as raw: version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0020_threads_meta_project_id" + assert version_row[0] == "0021_batch_acceptance" # Sanity: the invariant the index enforces now holds — at most one # active row per task_id. diff --git a/backend/tests/test_migration_0015_scheduled_task_enqueue.py b/backend/tests/test_migration_0015_scheduled_task_enqueue.py index a50cf1a6c..a9f998d8b 100644 --- a/backend/tests/test_migration_0015_scheduled_task_enqueue.py +++ b/backend/tests/test_migration_0015_scheduled_task_enqueue.py @@ -57,7 +57,7 @@ async def test_migration_interrupts_legacy_queue_and_adds_claim_fields(tmp_path: # Bootstrap always advances to the repository head after exercising # the 0015 migration behavior below. - assert version == "0020_threads_meta_project_id" + assert version == "0021_batch_acceptance" assert {"lease_owner", "lease_expires_at", "attempt_count"} <= columns.keys() assert columns["attempt_count"]["nullable"] is False assert overlap_policy == "enqueue" diff --git a/backend/tests/test_migration_0021_batch_acceptance.py b/backend/tests/test_migration_0021_batch_acceptance.py new file mode 100644 index 000000000..54db0bccd --- /dev/null +++ b/backend/tests/test_migration_0021_batch_acceptance.py @@ -0,0 +1,81 @@ +"""Upgrade legacy/project schemas, preserve rows, and reject missing batch fields.""" + +import asyncio + +import pytest +import sqlalchemy as sa +from alembic import command +from alembic.util.exc import CommandError +from sqlalchemy.ext.asyncio import create_async_engine + +from deerflow.persistence import bootstrap + + +@pytest.mark.asyncio +@pytest.mark.parametrize("source_revision", ["0018_oauth_identity_pg_partial", "0020_threads_meta_project_id"]) +async def test_upgrade_and_downgrade_preserve_legacy_batch_item(tmp_path, source_revision): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'batch.db'}") + cfg = bootstrap._get_alembic_config(engine) + try: + await asyncio.to_thread(bootstrap._upgrade, cfg, source_revision) + async with engine.begin() as conn: + await conn.execute( + sa.text( + "INSERT INTO subagent_batches (id,user_id,thread_id,submission_key,title,subagent_type,status,total_items,max_live_items,max_running_items,max_attempts,execution_spec,created_at,updated_at) " + "VALUES ('b','u','t','k','title','general-purpose','completed',1,1,1,2,'{}',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)" + ) + ) + if source_revision == "0020_threads_meta_project_id": + await conn.execute(sa.text("INSERT INTO projects (id,user_id,name,instructions,presentation,status,created_at,updated_at) VALUES ('p','u','existing project','','{}','active',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)")) + await conn.execute( + sa.text( + "INSERT INTO subagent_batch_items (id,batch_id,item_key,position,prompt,status,attempt,result,result_truncated,created_at,updated_at) " + "VALUES ('i','b','k',0,'p','succeeded',1,'old result',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)" + ) + ) + await bootstrap.bootstrap_schema(engine, backend="sqlite") + await bootstrap.bootstrap_schema(engine, backend="sqlite") + async with engine.connect() as conn: + columns = await conn.run_sync(lambda sync: {col["name"] for col in sa.inspect(sync).get_columns("subagent_batch_items")}) + assert {"acceptance_criteria", "acceptance_verdict"} <= columns + row = (await conn.execute(sa.text("SELECT result,status,acceptance_criteria,acceptance_verdict FROM subagent_batch_items"))).one() + assert tuple(row) == ("old result", "succeeded", None, None) + await asyncio.to_thread(command.downgrade, cfg, source_revision) + async with engine.connect() as conn: + columns = await conn.run_sync(lambda sync: {col["name"] for col in sa.inspect(sync).get_columns("subagent_batch_items")}) + assert "acceptance_verdict" not in columns + assert await conn.scalar(sa.text("SELECT result FROM subagent_batch_items")) == "old result" + if source_revision == "0020_threads_meta_project_id": + assert await conn.scalar(sa.text("SELECT name FROM projects WHERE id='p'")) == "existing project" + finally: + await engine.dispose() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("race", [False, True]) +async def test_forward_revision_cannot_skip_required_batch_columns(tmp_path, monkeypatch, race): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'forward.db'}") + cfg = bootstrap._get_alembic_config(engine) + try: + # Keep the project schema present so only the batch-column guard can + # reject this database, on both direct and concurrent-startup paths. + await asyncio.to_thread(bootstrap._upgrade, cfg, "0020_threads_meta_project_id") + if race: + + def raced_upgrade(*args): + sync = sa.create_engine(f"sqlite:///{tmp_path / 'forward.db'}") + try: + with sync.begin() as conn: + conn.execute(sa.text("UPDATE alembic_version SET version_num='0019_thread_incarnations'")) + finally: + sync.dispose() + raise CommandError("another deployment migrated first") + + monkeypatch.setattr(bootstrap, "_upgrade", raced_upgrade) + else: + async with engine.begin() as conn: + await conn.execute(sa.text("UPDATE alembic_version SET version_num='0019_thread_incarnations'")) + with pytest.raises(RuntimeError, match="missing required local schema: subagent_batch_items.acceptance_criteria, subagent_batch_items.acceptance_verdict"): + await bootstrap.bootstrap_schema(engine, backend="sqlite") + finally: + await engine.dispose() diff --git a/backend/tests/test_persistence_bootstrap.py b/backend/tests/test_persistence_bootstrap.py index ea365d5d6..cefeb2076 100644 --- a/backend/tests/test_persistence_bootstrap.py +++ b/backend/tests/test_persistence_bootstrap.py @@ -48,7 +48,7 @@ from deerflow.persistence.migrations._helpers import _normalize_default asyncio_test = pytest.mark.asyncio -HEAD = "0020_threads_meta_project_id" +HEAD = "0021_batch_acceptance" BASELINE = "0001_baseline" diff --git a/backend/tests/test_persistence_bootstrap_concurrency.py b/backend/tests/test_persistence_bootstrap_concurrency.py index 0c77aec51..4fb604e39 100644 --- a/backend/tests/test_persistence_bootstrap_concurrency.py +++ b/backend/tests/test_persistence_bootstrap_concurrency.py @@ -28,7 +28,7 @@ from deerflow.persistence.bootstrap import bootstrap_schema pytestmark = pytest.mark.asyncio -HEAD = "0020_threads_meta_project_id" +HEAD = "0021_batch_acceptance" def _url(tmp_path: Path) -> str: diff --git a/backend/tests/test_persistence_bootstrap_regression.py b/backend/tests/test_persistence_bootstrap_regression.py index da2a4d1ca..7d261d9ca 100644 --- a/backend/tests/test_persistence_bootstrap_regression.py +++ b/backend/tests/test_persistence_bootstrap_regression.py @@ -76,7 +76,7 @@ async def test_legacy_database_recovers_token_usage_column(tmp_path: Path) -> No cols = {row[1] for row in raw.execute("PRAGMA table_info(runs)").fetchall()} assert "token_usage_by_model" in cols version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0020_threads_meta_project_id" + assert version_row[0] == "0021_batch_acceptance" # And the read path that originally 500'd must now succeed. sf = get_session_factory() @@ -116,6 +116,6 @@ async def test_legacy_database_with_manual_alter_still_bootstraps(tmp_path: Path # No duplicate column -- list, not set, to catch dupes. assert cols.count("token_usage_by_model") == 1 version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0020_threads_meta_project_id" + assert version_row[0] == "0021_batch_acceptance" finally: await close_engine() diff --git a/backend/tests/test_persistence_forward_revision_compat.py b/backend/tests/test_persistence_forward_revision_compat.py index 36208250a..56806d9f6 100644 --- a/backend/tests/test_persistence_forward_revision_compat.py +++ b/backend/tests/test_persistence_forward_revision_compat.py @@ -29,7 +29,7 @@ from deerflow.persistence.engine import close_engine, get_engine, init_engine_fr from deerflow.persistence.mcp_tasks import McpTaskRepository from deerflow.persistence.thread_meta import ThreadMetaRepository -HEAD = "0020_threads_meta_project_id" +HEAD = "0021_batch_acceptance" POSTGRES_URL = os.environ.get("TEST_POSTGRES_URI")