[autofix.ci] apply automated fixes

Adds end-to-end coverage that exercises the §6 stack against real
``dify-db-1`` Postgres + real opendal storage — proves the I/O boundary
(DB row resolution + storage object read + ExtractProcessor invocation +
executor decision chain) actually works. Unit tests already cover the
parsing / truncation / orchestration logic; this suite catches regressions
where the loader signature drifts from
``ExtractProcessor.load_from_upload_file`` or where ``upload_files`` schema
expectations diverge.

9 scenarios covered:

UploadFileContentLoader (real DB + storage):
- Extracts text content for a seeded ``.txt`` row + storage object
- Rejects cross-tenant access without leaking content
- Short-circuits unsupported extensions (image) without touching storage
- Returns ``None`` for unknown file_id without raising

FileOutputCheckExecutor (real loader + stubbed model invoker):
- PASS verdict round-trips end-to-end with token usage propagated
- FAIL verdict round-trips end-to-end
- SKIPPED with ``produced_file_missing`` when produced UUID doesn't exist
- SKIPPED with ``benchmark_file_not_accessible`` on cross-tenant benchmark
- SKIPPED with ``output_check_model_error`` when invoker raises

The model invoker is the only piece stubbed; everything else is the real
production code path. Pattern follows
``test_remove_app_and_related_data_task``: seed via
``session_factory.create_session()`` with explicit commit, exercise, clean
up by ID + storage delete on teardown.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

api/core/workflow/node_factory.py

@@ -473,8 +473,11 @@ class DifyNodeFactory(NodeFactory):
             from clients.agent_backend import AgentBackendRunEventAdapter, AgentBackendRunRequestBuilder
             from clients.agent_backend.factory import create_agent_backend_run_client
             from core.workflow.nodes.agent_v2.file_tenant_validator import UploadFileTenantValidator
+            from core.workflow.nodes.agent_v2.output_check_executor import FileOutputCheckExecutor
+            from core.workflow.nodes.agent_v2.output_check_model_invoker import ModelRuntimeOutputCheckInvoker
             from core.workflow.nodes.agent_v2.output_failure_orchestrator import OutputFailureOrchestrator
             from core.workflow.nodes.agent_v2.output_type_checker import PerOutputTypeChecker
+            from core.workflow.nodes.agent_v2.upload_file_content_loader import UploadFileContentLoader
 
             return {
                 "binding_resolver": WorkflowAgentBindingResolver(),
@@ -489,10 +492,15 @@ class DifyNodeFactory(NodeFactory):
                 ),
                 "event_adapter": AgentBackendRunEventAdapter(),
                 "output_adapter": WorkflowAgentOutputAdapter(),
-                # Stage 4 §5/§7: per-output validation + failure orchestration. The
-                # tenant validator queries upload_files so it stays cheap when
-                # outputs contain no file refs.
+                # Stage 4 §5/§6/§7: per-output validation + benchmark check +
+                # failure orchestration. The tenant validator and content
+                # loader query upload_files lazily so they stay cheap when
+                # declared outputs include no file refs.
                 "type_checker": PerOutputTypeChecker(file_validator=UploadFileTenantValidator()),
+                "output_check_executor": FileOutputCheckExecutor(
+                    content_loader=UploadFileContentLoader(),
+                    model_invoker=ModelRuntimeOutputCheckInvoker(),
+                ),
                 "failure_orchestrator": OutputFailureOrchestrator(),
             }
         return {

api/core/workflow/nodes/agent_v2/agent_node.py

@@ -23,11 +23,12 @@ from core.workflow.system_variables import SystemVariableKey, get_system_text
 from graphon.enums import BuiltinNodeTypes, WorkflowNodeExecutionMetadataKey, WorkflowNodeExecutionStatus
 from graphon.node_events import NodeEventBase, NodeRunResult, StreamCompletedEvent
 from graphon.nodes.base.node import Node
-from models.agent_config_entities import WorkflowNodeJobConfig
+from models.agent_config_entities import AgentSoulConfig, AgentSoulModelConfig, WorkflowNodeJobConfig
 
 from .binding_resolver import WorkflowAgentBindingError, WorkflowAgentBindingResolver
 from .entities import DifyAgentNodeData
 from .output_adapter import WorkflowAgentOutputAdapter
+from .output_check_executor import FileOutputCheckExecutor, FileOutputCheckOutcome
 from .output_failure_orchestrator import (
     FailedOutput,
     OutputFailureDecision,
@@ -73,6 +74,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
         event_adapter: AgentBackendRunEventAdapter,
         output_adapter: WorkflowAgentOutputAdapter,
         type_checker: PerOutputTypeChecker,
+        output_check_executor: FileOutputCheckExecutor,
         failure_orchestrator: OutputFailureOrchestrator,
     ) -> None:
         super().__init__(
@@ -87,6 +89,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
         self._event_adapter = event_adapter
         self._output_adapter = output_adapter
         self._type_checker = type_checker
+        self._output_check_executor = output_check_executor
         self._failure_orchestrator = failure_orchestrator
 
     @classmethod
@@ -143,6 +146,22 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
         )
         outputs_by_name = {o.name: o for o in effective_outputs}
 
+        # Stage 4 §6: output check borrows the Agent Soul's model identity for
+        # its evaluator call. ``runtime_request_builder.build`` would also
+        # reject a missing model later, but we surface a deterministic error
+        # here so the failure_event has a sensible code.
+        agent_soul = AgentSoulConfig.model_validate(bundle.snapshot.config_snapshot_dict)
+        if agent_soul.model is None:
+            yield self._failure_event(
+                inputs=inputs,
+                process_data=process_data,
+                metadata=metadata,
+                error="Workflow Agent node requires Agent Soul model config.",
+                error_type="agent_model_not_configured",
+            )
+            return
+        agent_model: AgentSoulModelConfig = agent_soul.model
+
         # ──── Retry loop (Stage 4 §7) ────
         attempt = 0
         while True:
@@ -234,7 +253,7 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 )
                 return
 
-            # ──── Stage 4: per-output type check ────
+            # ──── Stage 4 §5: per-output type check ────
             type_check = self._type_checker.check(
                 declared_outputs=effective_outputs,
                 raw_output=terminal_event.output,
@@ -242,19 +261,38 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
             )
             self._record_type_check_metadata(metadata, type_check)
 
+            # ──── Stage 4 §6: file benchmark output check ────
+            # Only run when type check passes; comparing content of a value
+            # that's already mis-typed is wasted work and would produce a
+            # confusing second failure for the same root cause.
+            output_check: FileOutputCheckOutcome | None = None
             if not type_check.has_failures:
-                yield StreamCompletedEvent(
-                    node_run_result=self._output_adapter.build_success_result(
-                        event=terminal_event,
-                        inputs=inputs,
-                        process_data=process_data,
-                        metadata=metadata,
-                    )
+                output_check = self._output_check_executor.check_all(
+                    declared_outputs=effective_outputs,
+                    raw_output=terminal_event.output,
+                    tenant_id=dify_ctx.tenant_id,
+                    model_provider=agent_model.model_provider,
+                    model_name=agent_model.model,
+                    model_settings=agent_model.model_settings,
                 )
-                return
+                self._record_output_check_metadata(metadata, output_check)
 
-            # ──── Stage 4: orchestrate retry / default / fail ────
-            failures = [
+                if not output_check.has_failures:
+                    yield StreamCompletedEvent(
+                        node_run_result=self._output_adapter.build_success_result(
+                            event=terminal_event,
+                            inputs=inputs,
+                            process_data=process_data,
+                            metadata=metadata,
+                        )
+                    )
+                    return
+
+            # ──── Stage 4 §7: orchestrate retry / default / fail ────
+            # Aggregate failures from both stages. They are mutually exclusive
+            # (§6 only runs when §5 passes), so the resulting list contains
+            # exclusively TYPE_CHECK or OUTPUT_CHECK failures, never both.
+            failures: list[FailedOutput] = [
                 FailedOutput(
                     declared=outputs_by_name[result.name],
                     failure_kind=OutputFailureKind.TYPE_CHECK,
@@ -263,6 +301,17 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 for result in type_check.failures
                 if result.name in outputs_by_name
             ]
+            if output_check is not None:
+                failures.extend(
+                    FailedOutput(
+                        declared=outputs_by_name[result.output_name],
+                        failure_kind=OutputFailureKind.OUTPUT_CHECK,
+                        reason=result.reason,
+                    )
+                    for result in output_check.failures
+                    if result.output_name in outputs_by_name
+                )
+
             outcome = self._failure_orchestrator.decide(failures=failures, current_attempt=attempt)
             metadata["output_failure_decision"] = outcome.decision.value
             metadata["output_failure_reason"] = outcome.primary_reason
@@ -283,11 +332,17 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
                 )
                 return
 
-            error_type = (
-                "output_type_check_failed_fail_branch"
-                if outcome.decision == OutputFailureDecision.TAKE_FAIL_BRANCH
+            # Pick an error_type that reflects which stage produced the
+            # surviving failure(s); FAIL_BRANCH gets a suffixed variant so
+            # downstream metrics can tell the two paths apart.
+            base_code = (
+                "output_content_check_failed"
+                if OutputFailureKind.OUTPUT_CHECK in outcome.failure_kinds
                 else "output_type_check_failed"
             )
+            error_type = (
+                f"{base_code}_fail_branch" if outcome.decision == OutputFailureDecision.TAKE_FAIL_BRANCH else base_code
+            )
             yield self._failure_event(
                 inputs=inputs,
                 process_data=process_data,
@@ -384,6 +439,45 @@ class DifyAgentNode(Node[DifyAgentNodeData]):
             ],
         }
 
+    @staticmethod
+    def _record_output_check_metadata(metadata: dict[str, Any], outcome: FileOutputCheckOutcome) -> None:
+        """Persist §6 results into node metadata, including the D-2 usage bucket.
+
+        ``output_check_usage`` is keyed by output name so multiple file
+        outputs can share metadata without colliding. The bucket is recorded
+        even on FAILED / SKIPPED results so the billing pipeline observes any
+        model usage the executor did spend.
+        """
+        if not outcome.results:
+            return
+        per_output_usage: dict[str, dict[str, Any]] = {}
+        result_payload: list[dict[str, Any]] = []
+        for r in outcome.results:
+            result_payload.append(
+                {
+                    "name": r.output_name,
+                    "status": r.status.value,
+                    "reason": r.reason,
+                    "skip_reason": r.skip_reason.value if r.skip_reason else None,
+                    "content_truncated": r.content_truncated,
+                }
+            )
+            per_output_usage[r.output_name] = {
+                "prompt_tokens": r.usage.prompt_tokens,
+                "completion_tokens": r.usage.completion_tokens,
+                "total_tokens": r.usage.total_tokens,
+                "total_price": str(r.usage.total_price),
+                "currency": r.usage.currency,
+                "latency_ms": r.usage.latency_ms,
+            }
+        metadata["output_check"] = {
+            "passed": not outcome.has_failures,
+            "results": result_payload,
+        }
+        # D-2: keep this bucket separate from ``agent_run_usage`` so billing
+        # can tell agent inference apart from output verification.
+        metadata["output_check_usage"] = per_output_usage
+
     @staticmethod
     def _patch_event_with_defaults(
         event: AgentBackendRunSucceededInternalEvent,

api/core/workflow/nodes/agent_v2/output_check_executor.py

@@ -0,0 +1,455 @@
+"""Per-output file benchmark check executor for Workflow Agent Node v2.
+
+Stage 4 §6: after :class:`PerOutputTypeChecker` has confirmed that every
+declared output is structurally well-formed, this executor runs an *optional*,
+model-based semantic check on file outputs whose
+``DeclaredOutputCheckConfig.enabled`` is ``True``. The check is performed:
+
+* by **directly invoking the configured model** (NOT through the Agent backend),
+  because the backend's ``dify.output`` layer only enforces structural JSON
+  schema and has no notion of "compare two file payloads";
+* with **its token usage bucketed separately** as ``output_check_usage`` so
+  billing / observability never confuses agent-run usage with output-validation
+  usage (decision D-2);
+* with **file content loading and model invocation pluggable** via
+  :class:`FileContentLoader` and :class:`OutputCheckModelInvoker` Protocols so
+  unit tests can drive the executor without DB / network access.
+
+Failures here surface upward as
+``OutputFailureKind.OUTPUT_CHECK`` and feed the existing
+:class:`OutputFailureOrchestrator` decision chain alongside type-check
+failures.
+"""
+
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from decimal import Decimal
+from enum import StrEnum
+from typing import Any, Literal, Protocol
+
+from models.agent_config_entities import DeclaredOutputConfig, DeclaredOutputType
+
+
+class OutputCheckModelInvocationError(Exception):
+    """Raised by :class:`OutputCheckModelInvoker` when the LLM call fails.
+
+    The executor catches this and produces a ``SKIPPED`` result tagged with
+    :attr:`FileOutputCheckSkipReason.MODEL_INVOCATION_ERROR` so the surrounding
+    retry / fail-branch logic can still proceed deterministically.
+    """
+
+
+class FileOutputCheckStatus(StrEnum):
+    """Lifecycle status of a single file output after the benchmark check."""
+
+    PASSED = "passed"
+    FAILED = "failed"
+    # Check did not produce a pass/fail signal (unsupported file type, file
+    # not accessible, model error, ...). Skipped checks do NOT feed the
+    # failure orchestrator — they are surfaced as warnings in metadata.
+    SKIPPED = "skipped"
+
+
+class FileOutputCheckSkipReason(StrEnum):
+    """Why an output-check result is :attr:`FileOutputCheckStatus.SKIPPED`."""
+
+    UNSUPPORTED_FILE_FOR_OUTPUT_CHECK = "unsupported_file_for_output_check"
+    BENCHMARK_FILE_NOT_ACCESSIBLE = "benchmark_file_not_accessible"
+    PRODUCED_FILE_MISSING = "produced_file_missing"
+    MODEL_INVOCATION_ERROR = "output_check_model_error"
+
+
+@dataclass(frozen=True, slots=True)
+class FileOutputCheckUsage:
+    """Token / cost accounting for one output-check LLM invocation (§6.2 D-2).
+
+    Shape intentionally mirrors ``LLMUsage`` so future code can aggregate
+    multiple per-output usages and serialize them next to (but separate from)
+    agent run usage. Zero-valued instance means "no model call was made".
+    """
+
+    prompt_tokens: int = 0
+    completion_tokens: int = 0
+    total_tokens: int = 0
+    total_price: Decimal = field(default_factory=lambda: Decimal(0))
+    currency: str = "USD"
+    latency_ms: int = 0
+
+    def __add__(self, other: FileOutputCheckUsage) -> FileOutputCheckUsage:
+        if not isinstance(other, FileOutputCheckUsage):
+            return NotImplemented
+        return FileOutputCheckUsage(
+            prompt_tokens=self.prompt_tokens + other.prompt_tokens,
+            completion_tokens=self.completion_tokens + other.completion_tokens,
+            total_tokens=self.total_tokens + other.total_tokens,
+            total_price=self.total_price + other.total_price,
+            currency=self.currency if self.total_price else other.currency,
+            latency_ms=self.latency_ms + other.latency_ms,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class FileOutputCheckResult:
+    """Outcome of running benchmark check on one declared file output."""
+
+    output_name: str
+    status: FileOutputCheckStatus
+    reason: str
+    usage: FileOutputCheckUsage = field(default_factory=FileOutputCheckUsage)
+    skip_reason: FileOutputCheckSkipReason | None = None
+    content_truncated: bool = False
+
+
+@dataclass(frozen=True, slots=True)
+class FileOutputCheckOutcome:
+    """Aggregate of per-output check results for one agent backend run."""
+
+    results: tuple[FileOutputCheckResult, ...]
+
+    @property
+    def failures(self) -> tuple[FileOutputCheckResult, ...]:
+        return tuple(r for r in self.results if r.status == FileOutputCheckStatus.FAILED)
+
+    @property
+    def has_failures(self) -> bool:
+        return bool(self.failures)
+
+    @property
+    def total_usage(self) -> FileOutputCheckUsage:
+        total = FileOutputCheckUsage()
+        for r in self.results:
+            total = total + r.usage
+        return total
+
+    def by_name(self) -> dict[str, FileOutputCheckResult]:
+        return {r.output_name: r for r in self.results}
+
+
+@dataclass(frozen=True, slots=True)
+class LoadedFileContent:
+    """Output of :class:`FileContentLoader`.
+
+    ``text`` is empty when ``unsupported`` is ``True``; callers must check the
+    flag before reading text. ``truncated`` indicates the loader had to drop
+    content to fit the configured budget.
+    """
+
+    text: str
+    truncated: bool = False
+    unsupported: bool = False
+
+
+class FileContentLoader(Protocol):
+    """Resolve a ``file_id`` for the given tenant into model-readable text.
+
+    Returning ``None`` signals the file is missing / cross-tenant / failed to
+    extract — the executor maps that to a SKIPPED result instead of failing
+    the whole node. Implementations must not raise on those cases; raising is
+    reserved for unexpected runtime errors.
+    """
+
+    def load(self, *, file_id: str, tenant_id: str) -> LoadedFileContent | None: ...
+
+
+@dataclass(frozen=True, slots=True)
+class OutputCheckModelResponse:
+    """LLM response wrapping the raw assistant text plus token usage."""
+
+    text: str
+    usage: FileOutputCheckUsage
+
+
+class OutputCheckModelInvoker(Protocol):
+    """Direct (non-streaming) LLM invocation for output check.
+
+    The contract is intentionally narrow: one prompt in, one assistant message
+    out. The Agent Soul's model identity is passed explicitly so the executor
+    stays agnostic of how callers resolve the agent's model config.
+    """
+
+    def invoke(
+        self,
+        *,
+        tenant_id: str,
+        model_provider: str,
+        model_name: str,
+        prompt: str,
+        model_settings: Mapping[str, Any] | None = None,
+    ) -> OutputCheckModelResponse: ...
+
+
+# Recognized aliases for the file id key in a produced file payload. Mirrors
+# :data:`output_type_checker._FILE_ID_KEYS` so both stages handle the same
+# field set.
+_FILE_ID_KEYS: tuple[str, ...] = ("file_id", "upload_file_id", "tool_file_id")
+
+# Verdict / reason parsing patterns. The prompt instructs the model to start
+# with ``VERDICT: PASS|FAIL`` followed by ``REASON: ...``; we tolerate
+# whitespace and case variations.
+_VERDICT_PATTERN = re.compile(r"VERDICT\s*:\s*(PASS|FAIL)", re.IGNORECASE)
+_REASON_PATTERN = re.compile(r"REASON\s*:\s*(.+)", re.IGNORECASE | re.DOTALL)
+
+_DEFAULT_MAX_CONTENT_CHARS = 32_000
+_TRUNCATION_NOTICE = "…[content truncated]"
+
+
+class FileOutputCheckExecutor:
+    """Run benchmark checks against every file output that opted in."""
+
+    def __init__(
+        self,
+        *,
+        content_loader: FileContentLoader,
+        model_invoker: OutputCheckModelInvoker,
+        max_content_chars: int = _DEFAULT_MAX_CONTENT_CHARS,
+    ) -> None:
+        self._content_loader = content_loader
+        self._model_invoker = model_invoker
+        self._max_content_chars = max_content_chars
+
+    def check_all(
+        self,
+        *,
+        declared_outputs: list[DeclaredOutputConfig],
+        raw_output: Mapping[str, Any] | Any,
+        tenant_id: str,
+        model_provider: str,
+        model_name: str,
+        model_settings: Mapping[str, Any] | None = None,
+    ) -> FileOutputCheckOutcome:
+        """Run benchmark checks for the file outputs that opted in.
+
+        ``raw_output`` matches the shape of ``run_succeeded.data.output`` —
+        normally a dict, but we widen the signature like
+        :meth:`PerOutputTypeChecker.check` so a misbehaving backend cannot
+        crash the executor. Non-mapping payloads yield an empty outcome
+        (type-check already surfaced the failure).
+
+        Skips:
+          - non-file declared outputs (silently — handled by type check)
+          - file outputs without ``check.enabled``
+          - file outputs whose produced value is missing (already flagged by
+            type check; we do not want to surface a duplicate signal here)
+        """
+        if not isinstance(raw_output, Mapping):
+            return FileOutputCheckOutcome(results=())
+
+        results: list[FileOutputCheckResult] = []
+        for declared in declared_outputs:
+            if declared.type != DeclaredOutputType.FILE:
+                continue
+            if not (declared.check and declared.check.enabled):
+                continue
+            produced_value = raw_output.get(declared.name)
+            if produced_value is None:
+                results.append(
+                    FileOutputCheckResult(
+                        output_name=declared.name,
+                        status=FileOutputCheckStatus.SKIPPED,
+                        reason=f"Produced value for {declared.name!r} is missing.",
+                        skip_reason=FileOutputCheckSkipReason.PRODUCED_FILE_MISSING,
+                    )
+                )
+                continue
+            results.append(
+                self._check_one(
+                    declared=declared,
+                    produced_value=produced_value,
+                    tenant_id=tenant_id,
+                    model_provider=model_provider,
+                    model_name=model_name,
+                    model_settings=model_settings,
+                )
+            )
+        return FileOutputCheckOutcome(results=tuple(results))
+
+    def _check_one(
+        self,
+        *,
+        declared: DeclaredOutputConfig,
+        produced_value: Any,
+        tenant_id: str,
+        model_provider: str,
+        model_name: str,
+        model_settings: Mapping[str, Any] | None,
+    ) -> FileOutputCheckResult:
+        # ``declared.check`` is guaranteed non-None by the caller; access via
+        # the model validator that ensured prompt + benchmark_file_ref exist.
+        assert declared.check is not None
+        assert declared.check.enabled
+        check = declared.check
+
+        produced_file_id = self._extract_file_id(produced_value)
+        if produced_file_id is None:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason="Produced value lacks a recognized file_id field.",
+                skip_reason=FileOutputCheckSkipReason.PRODUCED_FILE_MISSING,
+            )
+        bench_ref = check.benchmark_file_ref or {}
+        benchmark_file_id = self._extract_file_id(bench_ref)
+        if benchmark_file_id is None:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason="benchmark_file_ref is missing a recognized file_id field.",
+                skip_reason=FileOutputCheckSkipReason.BENCHMARK_FILE_NOT_ACCESSIBLE,
+            )
+
+        produced = self._content_loader.load(file_id=produced_file_id, tenant_id=tenant_id)
+        if produced is None:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason=f"Produced file {produced_file_id!r} is not accessible to tenant.",
+                skip_reason=FileOutputCheckSkipReason.PRODUCED_FILE_MISSING,
+            )
+        if produced.unsupported:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason="Produced file type is not supported for output check.",
+                skip_reason=FileOutputCheckSkipReason.UNSUPPORTED_FILE_FOR_OUTPUT_CHECK,
+            )
+
+        benchmark = self._content_loader.load(file_id=benchmark_file_id, tenant_id=tenant_id)
+        if benchmark is None:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason=f"Benchmark file {benchmark_file_id!r} is not accessible to tenant.",
+                skip_reason=FileOutputCheckSkipReason.BENCHMARK_FILE_NOT_ACCESSIBLE,
+            )
+        if benchmark.unsupported:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason="Benchmark file type is not supported for output check.",
+                skip_reason=FileOutputCheckSkipReason.UNSUPPORTED_FILE_FOR_OUTPUT_CHECK,
+            )
+
+        benchmark_text, produced_text, truncated = self._truncate_for_budget(
+            benchmark_text=benchmark.text, produced_text=produced.text
+        )
+        truncated = truncated or benchmark.truncated or produced.truncated
+
+        # ``check.prompt`` is guaranteed non-None by the model validator when
+        # enabled=True, but we coerce defensively for older records.
+        user_prompt = check.prompt or ""
+        prompt = self._build_prompt(
+            user_prompt=user_prompt,
+            benchmark_text=benchmark_text,
+            produced_text=produced_text,
+        )
+
+        try:
+            response = self._model_invoker.invoke(
+                tenant_id=tenant_id,
+                model_provider=model_provider,
+                model_name=model_name,
+                prompt=prompt,
+                model_settings=model_settings,
+            )
+        except OutputCheckModelInvocationError as exc:
+            return FileOutputCheckResult(
+                output_name=declared.name,
+                status=FileOutputCheckStatus.SKIPPED,
+                reason=f"Model invocation failed: {exc}",
+                skip_reason=FileOutputCheckSkipReason.MODEL_INVOCATION_ERROR,
+                content_truncated=truncated,
+            )
+
+        verdict, reason = self._parse_verdict(response.text)
+        if verdict == "pass":
+            status = FileOutputCheckStatus.PASSED
+        elif verdict == "fail":
+            status = FileOutputCheckStatus.FAILED
+        else:
+            # Indeterminate output. We treat it as FAIL so the orchestrator
+            # gets a real signal; the raw model text is included in the
+            # reason for debugging.
+            status = FileOutputCheckStatus.FAILED
+            reason = f"Indeterminate model response: {response.text.strip()[:300]}"
+
+        return FileOutputCheckResult(
+            output_name=declared.name,
+            status=status,
+            reason=reason,
+            usage=response.usage,
+            content_truncated=truncated,
+        )
+
+    def _truncate_for_budget(
+        self,
+        *,
+        benchmark_text: str,
+        produced_text: str,
+    ) -> tuple[str, str, bool]:
+        """Split the char budget equally between benchmark and produced.
+
+        Returns ``(benchmark_text, produced_text, truncated)``. Each half is
+        capped at ``max_content_chars // 2`` so a single huge document cannot
+        starve the other side.
+        """
+        half = self._max_content_chars // 2
+        truncated = False
+        if len(benchmark_text) > half:
+            benchmark_text = benchmark_text[:half] + _TRUNCATION_NOTICE
+            truncated = True
+        if len(produced_text) > half:
+            produced_text = produced_text[:half] + _TRUNCATION_NOTICE
+            truncated = True
+        return benchmark_text, produced_text, truncated
+
+    @staticmethod
+    def _build_prompt(*, user_prompt: str, benchmark_text: str, produced_text: str) -> str:
+        return (
+            "You are an output validator. The user has defined the following acceptance criteria:\n"
+            "<criteria>\n"
+            f"{user_prompt.strip()}\n"
+            "</criteria>\n\n"
+            "Below is the BENCHMARK file the produced output should be evaluated against:\n"
+            "<benchmark>\n"
+            f"{benchmark_text}\n"
+            "</benchmark>\n\n"
+            "Below is the PRODUCED file from the agent run:\n"
+            "<produced>\n"
+            f"{produced_text}\n"
+            "</produced>\n\n"
+            "Decide whether the PRODUCED file satisfies the criteria when compared to the BENCHMARK.\n"
+            "Respond strictly in this format on two lines:\n"
+            "VERDICT: PASS\n"
+            "REASON: <one-sentence explanation>\n"
+            "(or VERDICT: FAIL when the produced file does not meet the criteria)."
+        )
+
+    @staticmethod
+    def _parse_verdict(text: str) -> tuple[Literal["pass", "fail", "unknown"], str]:
+        verdict_match = _VERDICT_PATTERN.search(text)
+        reason_match = _REASON_PATTERN.search(text)
+        if verdict_match is None:
+            return "unknown", text.strip()[:300]
+        verdict_raw = verdict_match.group(1).lower()
+        verdict: Literal["pass", "fail", "unknown"] = "pass" if verdict_raw == "pass" else "fail"
+        if reason_match is not None:
+            reason = reason_match.group(1).strip()
+            # Reasons can run multi-line in some model outputs; first line is
+            # usually the salient one.
+            reason = reason.split("\n", 1)[0].strip()
+        else:
+            reason = "" if verdict == "pass" else "Model returned FAIL without a reason."
+        return verdict, reason
+
+    @staticmethod
+    def _extract_file_id(value: Any) -> str | None:
+        if not isinstance(value, Mapping):
+            return None
+        for key in _FILE_ID_KEYS:
+            candidate = value.get(key)
+            if isinstance(candidate, str) and candidate:
+                return candidate
+        return None

api/core/workflow/nodes/agent_v2/output_check_model_invoker.py

@@ -0,0 +1,133 @@
+"""Production :class:`OutputCheckModelInvoker` backed by ``ModelManager``.
+
+Stage 4 §6: the file-output check needs a direct, non-streaming LLM call that
+yields one assistant message plus token usage. Implementation choices:
+
+* **No agent backend hop.** This is a one-shot evaluation, not an agentic
+  loop. Going through the agent backend would conflate output-check usage
+  with agent-run usage and introduce unnecessary protocol surface.
+* **Reuse Agent Soul's model identity.** Callers supply ``provider`` and
+  ``model_name`` from the same :class:`AgentSoulModelConfig` the agent itself
+  uses; the check therefore inherits the tenant's existing model credentials
+  and configuration without a separate setup.
+* **Bucket usage separately.** Returned ``FileOutputCheckUsage`` is later
+  recorded under ``WorkflowNodeExecutionMetadata.output_check_usage`` per
+  decision D-2; never merged with agent-run usage.
+
+Any exception raised inside ``ModelInstance.invoke_llm`` (provider error,
+credential issue, network timeout, ...) is converted to a single
+:class:`OutputCheckModelInvocationError`. The executor catches that and emits
+a SKIPPED result tagged ``output_check_model_error`` so the surrounding retry
+/ fail-branch logic still proceeds deterministically.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from typing import Any
+
+from core.model_manager import ModelManager
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from graphon.model_runtime.entities.message_entities import UserPromptMessage
+from graphon.model_runtime.entities.model_entities import ModelType
+
+from .output_check_executor import (
+    FileOutputCheckUsage,
+    OutputCheckModelInvocationError,
+    OutputCheckModelResponse,
+)
+
+# Resolves a tenant id to a fresh ``ModelManager``. Defined as a Callable
+# alias rather than a Protocol class so plain functions injected by tests
+# (e.g. ``lambda _: stub``) satisfy the type without subclassing.
+ModelManagerFactory = Callable[[str], ModelManager]
+
+
+class ModelRuntimeOutputCheckInvoker:
+    """Direct LLM invocation via the existing model_runtime stack.
+
+    A fresh :class:`ModelManager` is built per invocation by default so
+    credential-cache staleness cannot leak across workflow runs. Tests can
+    inject their own ``model_manager_factory`` to avoid touching the provider
+    manager and DB.
+    """
+
+    def __init__(
+        self,
+        model_manager_factory: ModelManagerFactory | None = None,
+    ) -> None:
+        self._factory: ModelManagerFactory = model_manager_factory or _default_model_manager_factory
+
+    def invoke(
+        self,
+        *,
+        tenant_id: str,
+        model_provider: str,
+        model_name: str,
+        prompt: str,
+        model_settings: Mapping[str, Any] | None = None,
+    ) -> OutputCheckModelResponse:
+        try:
+            manager = self._factory(tenant_id)
+            model_instance = manager.get_model_instance(
+                tenant_id=tenant_id,
+                provider=model_provider,
+                model_type=ModelType.LLM,
+                model=model_name,
+            )
+            result = model_instance.invoke_llm(
+                prompt_messages=[UserPromptMessage(content=prompt)],
+                model_parameters=dict(model_settings or {}),
+                stream=False,
+            )
+        except Exception as exc:
+            raise OutputCheckModelInvocationError(str(exc)) from exc
+
+        if not isinstance(result, LLMResult):
+            # ``stream=False`` is documented to return LLMResult; if the
+            # provider implementation breaks that contract surface it through
+            # the same uniform error path rather than a cryptic AttributeError
+            # later.
+            raise OutputCheckModelInvocationError(
+                f"Expected LLMResult from non-streaming invoke, got {type(result).__name__}"
+            )
+
+        text = _flatten_assistant_text(result)
+        usage = _to_file_output_check_usage(result.usage)
+        return OutputCheckModelResponse(text=text, usage=usage)
+
+
+def _default_model_manager_factory(tenant_id: str) -> ModelManager:
+    return ModelManager.for_tenant(tenant_id)
+
+
+def _flatten_assistant_text(result: LLMResult) -> str:
+    """Extract a plain string from ``AssistantPromptMessage.content``.
+
+    The model runtime allows multimodal content lists; for output check we
+    only ever expect a text response, but defensive flattening prevents an
+    unexpected list payload from crashing the parser.
+    """
+    content = result.message.content
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts: list[str] = []
+        for piece in content:
+            piece_text = getattr(piece, "data", None) or getattr(piece, "text", None)
+            if isinstance(piece_text, str):
+                parts.append(piece_text)
+        return "\n".join(parts)
+    return ""
+
+
+def _to_file_output_check_usage(usage: LLMUsage) -> FileOutputCheckUsage:
+    """Project an :class:`LLMUsage` into the executor's narrower shape."""
+    return FileOutputCheckUsage(
+        prompt_tokens=usage.prompt_tokens,
+        completion_tokens=usage.completion_tokens,
+        total_tokens=usage.total_tokens,
+        total_price=usage.total_price,
+        currency=usage.currency,
+        latency_ms=int(usage.latency * 1000),
+    )

api/core/workflow/nodes/agent_v2/upload_file_content_loader.py

@@ -0,0 +1,149 @@
+"""Production :class:`FileContentLoader` backed by ``upload_files`` + storage.
+
+Stage 4 §6: the :class:`FileOutputCheckExecutor` needs to read both the
+benchmark file (operator-supplied) and the agent-produced file as text so the
+LLM evaluator can compare them. Both are stored in Dify's ``upload_files``
+table; this adapter:
+
+1. resolves the ``file_id`` to an ``UploadFile`` row inside the caller's tenant
+   (cross-tenant access returns ``None`` — never raises);
+2. classifies the file's extension as text-extractable or unsupported (image /
+   archive / audio / video / executable are all treated as unsupported until
+   the executor learns to feed vision input or downloads);
+3. delegates text extraction to :class:`ExtractProcessor.load_from_upload_file`
+   so PDFs / Word / CSV / Markdown / HTML / Text all reuse the existing RAG
+   pipeline rather than re-implementing decoders.
+
+Any extraction failure (corrupt file, unsupported encoding, ETL backend down)
+becomes a ``LoadedFileContent`` with ``unsupported=True`` instead of an
+exception so the executor can map it to a deterministic ``SKIPPED`` result.
+"""
+
+from __future__ import annotations
+
+import logging
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.exc import DataError, SQLAlchemyError
+
+from core.db.session_factory import session_factory
+from core.rag.extractor.extract_processor import ExtractProcessor
+from models.model import UploadFile
+
+from .output_check_executor import LoadedFileContent
+
+logger = logging.getLogger(__name__)
+
+
+# File extensions explicitly rejected because text extraction either does not
+# apply (images, audio, video) or yields no useful comparison material
+# (archives, executables). Anything outside this set falls through to
+# ``ExtractProcessor`` which has its own fallback to a text decoder.
+_UNSUPPORTED_EXTENSIONS: frozenset[str] = frozenset(
+    {
+        # Images — handled later by a vision-capable code path; deferred to
+        # stage 4.1 per design doc §6.3.
+        ".png",
+        ".jpg",
+        ".jpeg",
+        ".gif",
+        ".webp",
+        ".bmp",
+        ".tiff",
+        ".svg",
+        ".ico",
+        # Audio / video.
+        ".mp3",
+        ".wav",
+        ".ogg",
+        ".flac",
+        ".mp4",
+        ".m4a",
+        ".mov",
+        ".webm",
+        ".avi",
+        ".mkv",
+        # Archives.
+        ".zip",
+        ".tar",
+        ".gz",
+        ".bz2",
+        ".7z",
+        ".rar",
+        # Executables / native binaries.
+        ".exe",
+        ".dll",
+        ".so",
+        ".dylib",
+        ".bin",
+        ".dmg",
+    }
+)
+
+
+class UploadFileContentLoader:
+    """Resolve an ``upload_files`` row → extracted plain text content.
+
+    Returns ``None`` (not an exception) for unknown / cross-tenant / DB-error
+    cases so the executor can produce a deterministic SKIPPED result. Returns
+    a ``LoadedFileContent`` with ``unsupported=True`` when the file format is
+    recognized but not text-extractable (e.g. image, archive). Returns a
+    populated ``LoadedFileContent`` on success.
+    """
+
+    def load(self, *, file_id: str, tenant_id: str) -> LoadedFileContent | None:
+        if not file_id or not tenant_id:
+            return None
+        try:
+            UUID(file_id)
+        except (ValueError, TypeError):
+            return None
+        try:
+            with session_factory.create_session() as session:
+                upload_file = session.scalar(select(UploadFile).where(UploadFile.id == file_id))
+        except (DataError, SQLAlchemyError):
+            logger.warning("UploadFileContentLoader: DB error while resolving file_id=%s", file_id, exc_info=True)
+            return None
+
+        if upload_file is None or upload_file.tenant_id != tenant_id:
+            return None
+
+        extension = self._extension_of(upload_file)
+        if extension in _UNSUPPORTED_EXTENSIONS:
+            return LoadedFileContent(text="", unsupported=True)
+
+        try:
+            extracted = ExtractProcessor.load_from_upload_file(upload_file, return_text=True)
+        except Exception:
+            # Any failure inside the extraction pipeline (corrupt file,
+            # missing storage object, ETL backend down, plugin error...) is
+            # surfaced as "unsupported" so the executor produces a SKIPPED
+            # result rather than failing the whole node.
+            logger.warning(
+                "UploadFileContentLoader: extraction failed for file_id=%s ext=%s",
+                file_id,
+                extension,
+                exc_info=True,
+            )
+            return LoadedFileContent(text="", unsupported=True)
+
+        if not isinstance(extracted, str):
+            # ExtractProcessor.load_from_upload_file returns ``list[Document]``
+            # only when ``return_text=False``; defensive guard.
+            return LoadedFileContent(text="", unsupported=True)
+
+        return LoadedFileContent(text=extracted)
+
+    @staticmethod
+    def _extension_of(upload_file: UploadFile) -> str:
+        """Lowercased ``.ext`` form of :attr:`UploadFile.extension`.
+
+        ``UploadFile.extension`` is stored without the leading dot (e.g.
+        ``"pdf"``), but the unsupported-set is keyed by ``".pdf"`` form so
+        we normalize before lookup. Empty extension stays empty.
+        """
+        raw = (upload_file.extension or "").strip().lower()
+        if not raw:
+            return ""
+        return raw if raw.startswith(".") else f".{raw}"

api/tests/integration_tests/core/workflow/nodes/agent_v2/test_output_check_executor_e2e.py

@@ -0,0 +1,393 @@
+"""End-to-end tests for the §6 file-output benchmark check stack.
+
+Exercises ``UploadFileContentLoader`` against real ``upload_files`` rows +
+storage, then drives ``FileOutputCheckExecutor.check_all`` end-to-end with a
+stubbed model invoker so we never hit a real LLM. This proves the I/O
+boundary (DB + storage + ExtractProcessor) actually works — unit tests can't
+catch a regression where the loader signature drifts from
+``ExtractProcessor.load_from_upload_file`` or the executor's file_id
+resolution differs from the real ``upload_files`` schema.
+
+Pattern follows ``test_remove_app_and_related_data_task``: seed via
+``session_factory.create_session()`` with explicit commits + cleanup by ID
+on teardown.
+"""
+
+from __future__ import annotations
+
+import uuid
+from collections.abc import Generator
+from datetime import UTC, datetime
+from decimal import Decimal
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+from sqlalchemy import delete
+
+from core.db.session_factory import session_factory
+from core.workflow.nodes.agent_v2.output_check_executor import (
+    FileOutputCheckExecutor,
+    FileOutputCheckSkipReason,
+    FileOutputCheckStatus,
+    FileOutputCheckUsage,
+    OutputCheckModelInvocationError,
+    OutputCheckModelResponse,
+)
+from core.workflow.nodes.agent_v2.upload_file_content_loader import (
+    UploadFileContentLoader,
+)
+from extensions.ext_storage import storage
+from extensions.storage.storage_type import StorageType
+from models.enums import CreatorUserRole
+from models.model import UploadFile
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Fixture: real upload_files row + real storage object
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _make_upload_file(
+    *,
+    tenant_id: str,
+    created_by: str,
+    name: str,
+    extension: str,
+    content: bytes,
+) -> UploadFile:
+    """Build (without persisting) an ``UploadFile`` row + its storage key.
+
+    ``UploadFile.__init__`` allocates a fresh UUID for ``.id``; we then
+    derive the storage key from it so the row + the blob agree.
+    """
+    upload = UploadFile(
+        tenant_id=tenant_id,
+        storage_type=StorageType.OPENDAL,
+        key="placeholder",  # overwritten below once we know the generated id
+        name=name,
+        size=len(content),
+        extension=extension,
+        mime_type="text/plain",
+        created_by_role=CreatorUserRole.ACCOUNT,
+        created_by=created_by,
+        created_at=datetime.now(UTC),
+        used=False,
+    )
+    upload.key = f"upload_files/{tenant_id}/{upload.id}.{extension}"
+    return upload
+
+
+@pytest.fixture
+def stored_text_file(
+    flask_req_ctx,
+) -> Generator[tuple[UploadFile, str], None, None]:
+    """Seed one real ``upload_files`` row + write its content to storage.
+
+    Yields ``(upload_file, content_text)``. Cleans up DB row and the storage
+    object on teardown so we never leave debris.
+    """
+    tenant_id = str(uuid.uuid4())
+    created_by = str(uuid.uuid4())
+    content_text = "This is the benchmark contract.\nIt has multiple lines.\nLine 3."
+    upload = _make_upload_file(
+        tenant_id=tenant_id,
+        created_by=created_by,
+        name="benchmark.txt",
+        extension="txt",
+        content=content_text.encode("utf-8"),
+    )
+    storage.save(upload.key, content_text.encode("utf-8"))
+    with session_factory.create_session() as session:
+        session.add(upload)
+        session.commit()
+        upload_id, upload_key = upload.id, upload.key
+
+    try:
+        yield upload, content_text
+    finally:
+        try:
+            storage.delete(upload_key)
+        except Exception:
+            pass
+        with session_factory.create_session() as session:
+            session.execute(delete(UploadFile).where(UploadFile.id == upload_id))
+            session.commit()
+
+
+@pytest.fixture
+def two_stored_files(
+    flask_req_ctx,
+) -> Generator[tuple[UploadFile, UploadFile, str, str], None, None]:
+    """Two real ``upload_files`` rows — produced + benchmark — for executor
+    end-to-end coverage."""
+    tenant_id = str(uuid.uuid4())
+    created_by = str(uuid.uuid4())
+
+    benchmark_text = "Required sections: introduction, methodology, results, conclusion."
+    produced_text = "introduction\n\nmethodology\n\nresults\n\nconclusion\n\nsign-off"
+
+    benchmark = _make_upload_file(
+        tenant_id=tenant_id,
+        created_by=created_by,
+        name="benchmark.md",
+        extension="md",
+        content=benchmark_text.encode("utf-8"),
+    )
+    produced = _make_upload_file(
+        tenant_id=tenant_id,
+        created_by=created_by,
+        name="produced.md",
+        extension="md",
+        content=produced_text.encode("utf-8"),
+    )
+
+    storage.save(benchmark.key, benchmark_text.encode("utf-8"))
+    storage.save(produced.key, produced_text.encode("utf-8"))
+    with session_factory.create_session() as session:
+        session.add(benchmark)
+        session.add(produced)
+        session.commit()
+        keys = [benchmark.key, produced.key]
+        ids = [benchmark.id, produced.id]
+
+    try:
+        yield benchmark, produced, benchmark_text, produced_text
+    finally:
+        for k in keys:
+            try:
+                storage.delete(k)
+            except Exception:
+                pass
+        with session_factory.create_session() as session:
+            session.execute(delete(UploadFile).where(UploadFile.id.in_(ids)))
+            session.commit()
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# UploadFileContentLoader against real DB + real storage
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_upload_file_content_loader_extracts_text_for_real_upload_file(stored_text_file):
+    loader = UploadFileContentLoader()
+    upload, content_text = stored_text_file
+    loaded = loader.load(file_id=upload.id, tenant_id=upload.tenant_id)
+    assert loaded is not None
+    assert loaded.unsupported is False
+    # ExtractProcessor for ``.txt`` uses TextExtractor; content should round-trip.
+    assert content_text.split("\n")[0] in loaded.text
+
+
+def test_upload_file_content_loader_rejects_cross_tenant_access(stored_text_file):
+    loader = UploadFileContentLoader()
+    upload, _ = stored_text_file
+    intruder_tenant = str(uuid.uuid4())
+    assert loader.load(file_id=upload.id, tenant_id=intruder_tenant) is None
+
+
+def test_upload_file_content_loader_short_circuits_unsupported_extension(flask_req_ctx):
+    """Image extensions never reach storage; verify the short-circuit holds
+    against a real ``upload_files`` row."""
+    tenant_id = str(uuid.uuid4())
+    upload = _make_upload_file(
+        tenant_id=tenant_id,
+        created_by=str(uuid.uuid4()),
+        name="photo.png",
+        extension="png",
+        content=b"\x89PNG\r\n...",
+    )
+    with session_factory.create_session() as session:
+        session.add(upload)
+        session.commit()
+        upload_id = upload.id
+
+    try:
+        loader = UploadFileContentLoader()
+        loaded = loader.load(file_id=upload_id, tenant_id=tenant_id)
+        assert loaded is not None
+        assert loaded.unsupported is True
+        assert loaded.text == ""
+    finally:
+        with session_factory.create_session() as session:
+            session.execute(delete(UploadFile).where(UploadFile.id == upload_id))
+            session.commit()
+
+
+def test_upload_file_content_loader_handles_missing_file_id_gracefully():
+    """Unknown UUID just returns ``None`` — never raises."""
+    loader = UploadFileContentLoader()
+    assert loader.load(file_id=str(uuid.uuid4()), tenant_id=str(uuid.uuid4())) is None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# FileOutputCheckExecutor with real DB-backed loader + stubbed invoker
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _patch_extraction_to_text(monkey_text: str):
+    """Substitute ``ExtractProcessor.load_from_upload_file`` to a deterministic
+    string so the test does not rely on the ``unstructured`` / ETL stack being
+    fully wired in the integration env (the ``.md`` extractor calls into the
+    Markdown parser which may not be initialised here)."""
+    return patch(
+        "core.workflow.nodes.agent_v2.upload_file_content_loader.ExtractProcessor.load_from_upload_file",
+        return_value=monkey_text,
+    )
+
+
+class _PassingStubInvoker:
+    def __init__(self) -> None:
+        self.calls: list[dict[str, Any]] = []
+
+    def invoke(self, **kwargs: Any) -> OutputCheckModelResponse:
+        self.calls.append(kwargs)
+        return OutputCheckModelResponse(
+            text="VERDICT: PASS\nREASON: All required sections present.",
+            usage=FileOutputCheckUsage(
+                prompt_tokens=42,
+                completion_tokens=8,
+                total_tokens=50,
+                total_price=Decimal("0.001"),
+                latency_ms=120,
+            ),
+        )
+
+
+class _FailingStubInvoker:
+    def invoke(self, **_kwargs: Any) -> OutputCheckModelResponse:
+        return OutputCheckModelResponse(
+            text="VERDICT: FAIL\nREASON: Missing methodology section.",
+            usage=FileOutputCheckUsage(prompt_tokens=20, completion_tokens=5, total_tokens=25),
+        )
+
+
+class _RaisingStubInvoker:
+    def invoke(self, **_kwargs: Any) -> OutputCheckModelResponse:
+        raise OutputCheckModelInvocationError("simulated provider outage")
+
+
+def _declared_file_output(name: str, benchmark_file_id: str):
+    """Construct a declared output that opts into output check, pointing at
+    the supplied benchmark file id."""
+    from models.agent_config_entities import (
+        DeclaredOutputCheckConfig,
+        DeclaredOutputConfig,
+        DeclaredOutputType,
+    )
+
+    return DeclaredOutputConfig(
+        name=name,
+        type=DeclaredOutputType.FILE,
+        check=DeclaredOutputCheckConfig(
+            enabled=True,
+            prompt="Verify the produced file matches the benchmark structure.",
+            benchmark_file_ref={"file_id": benchmark_file_id},
+        ),
+    )
+
+
+def test_executor_passes_through_real_db_when_model_says_pass(two_stored_files):
+    benchmark, produced, benchmark_text, produced_text = two_stored_files
+    invoker = _PassingStubInvoker()
+    executor = FileOutputCheckExecutor(
+        content_loader=UploadFileContentLoader(),
+        model_invoker=invoker,
+    )
+
+    with _patch_extraction_to_text("__loader_text__"):
+        outcome = executor.check_all(
+            declared_outputs=[_declared_file_output("report", benchmark.id)],
+            raw_output={"report": {"file_id": produced.id}},
+            tenant_id=benchmark.tenant_id,
+            model_provider="openai",
+            model_name="gpt-4",
+        )
+
+    assert len(outcome.results) == 1
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.PASSED
+    assert result.reason == "All required sections present."
+    assert result.usage.total_tokens == 50
+    # Usage should be propagated end-to-end; verify the aggregate too.
+    assert outcome.total_usage.total_tokens == 50
+    # The invoker received a prompt that embeds both files' content.
+    assert "__loader_text__" in invoker.calls[0]["prompt"]
+
+
+def test_executor_fails_through_real_db_when_model_says_fail(two_stored_files):
+    benchmark, produced, _, _ = two_stored_files
+    executor = FileOutputCheckExecutor(
+        content_loader=UploadFileContentLoader(),
+        model_invoker=_FailingStubInvoker(),
+    )
+    with _patch_extraction_to_text("__loader_text__"):
+        outcome = executor.check_all(
+            declared_outputs=[_declared_file_output("report", benchmark.id)],
+            raw_output={"report": {"file_id": produced.id}},
+            tenant_id=benchmark.tenant_id,
+            model_provider="openai",
+            model_name="gpt-4",
+        )
+    assert outcome.has_failures
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.FAILED
+    assert "Missing methodology" in result.reason
+
+
+def test_executor_skips_when_produced_file_inaccessible(two_stored_files):
+    """Produced file id doesn't exist in DB → SKIPPED with PRODUCED_FILE_MISSING."""
+    benchmark, _, _, _ = two_stored_files
+    executor = FileOutputCheckExecutor(
+        content_loader=UploadFileContentLoader(),
+        model_invoker=_PassingStubInvoker(),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_declared_file_output("report", benchmark.id)],
+        raw_output={"report": {"file_id": str(uuid.uuid4())}},  # phantom file
+        tenant_id=benchmark.tenant_id,
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.SKIPPED
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.PRODUCED_FILE_MISSING
+
+
+def test_executor_skips_when_benchmark_file_cross_tenant(two_stored_files):
+    """Benchmark file owned by another tenant → SKIPPED with BENCHMARK_FILE_NOT_ACCESSIBLE."""
+    benchmark, produced, _, _ = two_stored_files
+    executor = FileOutputCheckExecutor(
+        content_loader=UploadFileContentLoader(),
+        model_invoker=_PassingStubInvoker(),
+    )
+    intruder_tenant = str(uuid.uuid4())
+    outcome = executor.check_all(
+        declared_outputs=[_declared_file_output("report", benchmark.id)],
+        raw_output={"report": {"file_id": produced.id}},
+        tenant_id=intruder_tenant,  # mismatch → loader returns None for both files
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    # Either produced or benchmark resolves first to None; either way the
+    # result is a SKIPPED row.
+    assert outcome.results[0].status == FileOutputCheckStatus.SKIPPED
+
+
+def test_executor_skips_when_model_invoker_raises(two_stored_files):
+    """Provider outage → SKIPPED with MODEL_INVOCATION_ERROR."""
+    benchmark, produced, _, _ = two_stored_files
+    executor = FileOutputCheckExecutor(
+        content_loader=UploadFileContentLoader(),
+        model_invoker=_RaisingStubInvoker(),
+    )
+    with _patch_extraction_to_text("__loader_text__"):
+        outcome = executor.check_all(
+            declared_outputs=[_declared_file_output("report", benchmark.id)],
+            raw_output={"report": {"file_id": produced.id}},
+            tenant_id=benchmark.tenant_id,
+            model_provider="openai",
+            model_name="gpt-4",
+        )
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.SKIPPED
+    assert result.skip_reason == FileOutputCheckSkipReason.MODEL_INVOCATION_ERROR
+    assert "simulated provider outage" in result.reason

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_agent_node.py

@@ -99,6 +99,14 @@ def _node(*, scenario: FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCE
         },
         call_depth=0,
     )
+    from core.workflow.nodes.agent_v2.output_check_executor import (
+        FileOutputCheckExecutor,
+        LoadedFileContent,
+        OutputCheckModelResponse,
+    )
+    from core.workflow.nodes.agent_v2.output_check_executor import (
+        FileOutputCheckUsage as _CheckUsage,
+    )
     from core.workflow.nodes.agent_v2.output_failure_orchestrator import OutputFailureOrchestrator
     from core.workflow.nodes.agent_v2.output_type_checker import PerOutputTypeChecker
 
@@ -106,6 +114,16 @@ def _node(*, scenario: FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCE
         def is_owned_by_tenant(self, *, file_id: str, tenant_id: str) -> bool:
             return True
 
+    class _NeverInvokedContentLoader:
+        """Default fakes never enable an output check, so this loader is unused."""
+
+        def load(self, *, file_id: str, tenant_id: str):  # pragma: no cover - defensive
+            return LoadedFileContent(text="", unsupported=True)
+
+    class _NeverInvokedModelInvoker:
+        def invoke(self, **_kwargs):  # pragma: no cover - defensive
+            return OutputCheckModelResponse(text="VERDICT: PASS", usage=_CheckUsage())
+
     return DifyAgentNode(
         node_id="agent-node",
         data=DifyAgentNodeData.model_validate({"type": BuiltinNodeTypes.AGENT, "version": "2"}),
@@ -117,6 +135,10 @@ def _node(*, scenario: FakeAgentBackendScenario = FakeAgentBackendScenario.SUCCE
         event_adapter=AgentBackendRunEventAdapter(),
         output_adapter=WorkflowAgentOutputAdapter(),
         type_checker=PerOutputTypeChecker(file_validator=_AlwaysAllowFileValidator()),
+        output_check_executor=FileOutputCheckExecutor(
+            content_loader=_NeverInvokedContentLoader(),
+            model_invoker=_NeverInvokedModelInvoker(),
+        ),
         failure_orchestrator=OutputFailureOrchestrator(),
     )
 

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_output_check_executor.py

@@ -0,0 +1,672 @@
+"""Unit tests for FileOutputCheckExecutor.
+
+Stage 4 §6. The executor orchestrates per-file benchmark checks; the loader
+and model invoker are injected as Protocols so these tests stay free of DB
+and network access.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from decimal import Decimal
+from typing import Any
+
+import pytest
+
+from core.workflow.nodes.agent_v2.output_check_executor import (
+    FileContentLoader,
+    FileOutputCheckExecutor,
+    FileOutputCheckOutcome,
+    FileOutputCheckSkipReason,
+    FileOutputCheckStatus,
+    FileOutputCheckUsage,
+    LoadedFileContent,
+    OutputCheckModelInvocationError,
+    OutputCheckModelInvoker,
+    OutputCheckModelResponse,
+)
+from models.agent_config_entities import (
+    DeclaredOutputCheckConfig,
+    DeclaredOutputConfig,
+    DeclaredOutputType,
+)
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Stubs
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+class StubContentLoader(FileContentLoader):
+    """Maps ``file_id`` → ``LoadedFileContent`` (or ``None`` for not accessible)."""
+
+    def __init__(self, *, contents: Mapping[str, LoadedFileContent | None]) -> None:
+        self._contents = dict(contents)
+        self.calls: list[tuple[str, str]] = []
+
+    def load(self, *, file_id: str, tenant_id: str) -> LoadedFileContent | None:
+        self.calls.append((file_id, tenant_id))
+        return self._contents.get(file_id)
+
+
+class StubModelInvoker(OutputCheckModelInvoker):
+    """Returns a canned response or raises a configured error."""
+
+    def __init__(
+        self,
+        *,
+        response: OutputCheckModelResponse | None = None,
+        error: OutputCheckModelInvocationError | None = None,
+    ) -> None:
+        self._response = response
+        self._error = error
+        self.calls: list[Mapping[str, Any]] = []
+
+    def invoke(
+        self,
+        *,
+        tenant_id: str,
+        model_provider: str,
+        model_name: str,
+        prompt: str,
+        model_settings: Mapping[str, Any] | None = None,
+    ) -> OutputCheckModelResponse:
+        self.calls.append(
+            {
+                "tenant_id": tenant_id,
+                "model_provider": model_provider,
+                "model_name": model_name,
+                "prompt": prompt,
+                "model_settings": dict(model_settings or {}),
+            }
+        )
+        if self._error is not None:
+            raise self._error
+        assert self._response is not None
+        return self._response
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Helpers
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def _file_output(
+    *,
+    name: str = "report",
+    enabled: bool = True,
+    prompt: str | None = "Check structure matches benchmark.",
+    benchmark_ref: dict[str, Any] | None = None,
+) -> DeclaredOutputConfig:
+    if enabled:
+        # Validator requires a populated benchmark_file_ref when enabled. The
+        # default ``{"file_id": "bench-1"}`` matches the canonical contract;
+        # tests that want to exercise "missing file_id" pass a ref like
+        # ``{"filename": "x.pdf"}`` instead.
+        ref = benchmark_ref if benchmark_ref is not None else {"file_id": "bench-1"}
+        check = DeclaredOutputCheckConfig(
+            enabled=True,
+            prompt=prompt,
+            benchmark_file_ref=ref,
+        )
+    else:
+        check = DeclaredOutputCheckConfig(enabled=False)
+    return DeclaredOutputConfig(
+        name=name,
+        type=DeclaredOutputType.FILE,
+        check=check,
+    )
+
+
+def _ok_response(text: str, *, prompt_tokens: int = 10, completion_tokens: int = 5) -> OutputCheckModelResponse:
+    return OutputCheckModelResponse(
+        text=text,
+        usage=FileOutputCheckUsage(
+            prompt_tokens=prompt_tokens,
+            completion_tokens=completion_tokens,
+            total_tokens=prompt_tokens + completion_tokens,
+            total_price=Decimal("0.001"),
+            currency="USD",
+            latency_ms=42,
+        ),
+    )
+
+
+def _make_executor(
+    *,
+    contents: Mapping[str, LoadedFileContent | None] | None = None,
+    response: OutputCheckModelResponse | None = None,
+    error: OutputCheckModelInvocationError | None = None,
+    max_content_chars: int = 32_000,
+) -> tuple[FileOutputCheckExecutor, StubContentLoader, StubModelInvoker]:
+    loader = StubContentLoader(contents=contents or {})
+    invoker = StubModelInvoker(response=response, error=error)
+    executor = FileOutputCheckExecutor(
+        content_loader=loader,
+        model_invoker=invoker,
+        max_content_chars=max_content_chars,
+    )
+    return executor, loader, invoker
+
+
+def _ok_loader_contents() -> dict[str, LoadedFileContent]:
+    return {
+        "produced-1": LoadedFileContent(text="produced body"),
+        "bench-1": LoadedFileContent(text="benchmark body"),
+    }
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Skip-path: nothing-to-check declarations
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_non_file_outputs_are_silently_skipped():
+    executor, _, invoker = _make_executor(response=_ok_response("VERDICT: PASS\nREASON: ok"))
+    declared = [
+        DeclaredOutputConfig(name="text", type=DeclaredOutputType.STRING),
+        DeclaredOutputConfig(name="json", type=DeclaredOutputType.OBJECT),
+    ]
+    outcome = executor.check_all(
+        declared_outputs=declared,
+        raw_output={"text": "hi", "json": {}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results == ()
+    assert invoker.calls == []
+
+
+def test_file_output_without_check_enabled_is_skipped():
+    executor, _, invoker = _make_executor(response=_ok_response("VERDICT: PASS"))
+    declared = [_file_output(enabled=False)]
+    outcome = executor.check_all(
+        declared_outputs=declared,
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results == ()
+    assert invoker.calls == []
+
+
+def test_missing_produced_value_emits_skipped_result():
+    executor, _, invoker = _make_executor(response=_ok_response("VERDICT: PASS"))
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert len(outcome.results) == 1
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.SKIPPED
+    assert result.skip_reason == FileOutputCheckSkipReason.PRODUCED_FILE_MISSING
+    assert invoker.calls == []
+
+
+def test_produced_value_lacking_file_id_is_skipped():
+    executor, _, invoker = _make_executor(response=_ok_response("VERDICT: PASS"))
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"filename": "x.pdf"}},  # no file_id
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.SKIPPED
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.PRODUCED_FILE_MISSING
+    assert invoker.calls == []
+
+
+def test_benchmark_ref_lacking_file_id_is_skipped():
+    executor, _, invoker = _make_executor(response=_ok_response("VERDICT: PASS"))
+    outcome = executor.check_all(
+        # benchmark_file_ref is populated (validator allows it) but has no
+        # recognized file_id key — only metadata. Executor must SKIP, not
+        # crash.
+        declared_outputs=[_file_output(benchmark_ref={"filename": "ignored.pdf"})],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.SKIPPED
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.BENCHMARK_FILE_NOT_ACCESSIBLE
+    assert invoker.calls == []
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Loader-driven SKIP cases
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_produced_file_inaccessible_is_skipped():
+    executor, _, invoker = _make_executor(
+        contents={"produced-1": None, "bench-1": LoadedFileContent(text="b")},
+        response=_ok_response("VERDICT: PASS"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.PRODUCED_FILE_MISSING
+    assert invoker.calls == []
+
+
+def test_benchmark_file_inaccessible_is_skipped():
+    executor, _, invoker = _make_executor(
+        contents={"produced-1": LoadedFileContent(text="p"), "bench-1": None},
+        response=_ok_response("VERDICT: PASS"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.BENCHMARK_FILE_NOT_ACCESSIBLE
+    assert invoker.calls == []
+
+
+def test_produced_file_unsupported_is_skipped():
+    executor, _, invoker = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text="", unsupported=True),
+            "bench-1": LoadedFileContent(text="b"),
+        },
+        response=_ok_response("VERDICT: PASS"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.UNSUPPORTED_FILE_FOR_OUTPUT_CHECK
+    assert invoker.calls == []
+
+
+def test_benchmark_file_unsupported_is_skipped():
+    executor, _, invoker = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text="p"),
+            "bench-1": LoadedFileContent(text="", unsupported=True),
+        },
+        response=_ok_response("VERDICT: PASS"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].skip_reason == FileOutputCheckSkipReason.UNSUPPORTED_FILE_FOR_OUTPUT_CHECK
+    assert invoker.calls == []
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Model error
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_model_invocation_error_yields_skipped_with_usage_zero():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        error=OutputCheckModelInvocationError("provider down"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.SKIPPED
+    assert result.skip_reason == FileOutputCheckSkipReason.MODEL_INVOCATION_ERROR
+    assert "provider down" in result.reason
+    assert result.usage == FileOutputCheckUsage()  # no usage on failure
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Verdict parsing
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_pass_verdict_is_parsed_with_reason():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: PASS\nREASON: matches structure."),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.PASSED
+    assert result.reason == "matches structure."
+
+
+def test_fail_verdict_is_parsed_with_reason():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: FAIL\nREASON: missing section 3."),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.FAILED
+    assert result.reason == "missing section 3."
+
+
+def test_verdict_parsing_tolerates_case_and_whitespace():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("verdict:   pass   \n  reason  :  ok!"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.PASSED
+
+
+def test_indeterminate_response_is_treated_as_failure():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("Sure thing! The file looks great I guess."),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    result = outcome.results[0]
+    assert result.status == FileOutputCheckStatus.FAILED
+    assert "Indeterminate model response" in result.reason
+
+
+def test_pass_without_reason_yields_empty_reason():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: PASS"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.PASSED
+    assert outcome.results[0].reason == ""
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Truncation
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_text_within_budget_does_not_truncate():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+        max_content_chars=10_000,
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].content_truncated is False
+
+
+def test_oversized_content_is_truncated():
+    big_text = "x" * 50_000
+    executor, _, invoker = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text=big_text),
+            "bench-1": LoadedFileContent(text="ok"),
+        },
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+        max_content_chars=10_000,
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].content_truncated is True
+    # Prompt should contain the truncation marker because the produced body
+    # exceeded the half-budget.
+    assert "content truncated" in invoker.calls[0]["prompt"]
+
+
+def test_loader_reported_truncation_propagates_to_result():
+    executor, _, _ = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text="p", truncated=True),
+            "bench-1": LoadedFileContent(text="b"),
+        },
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].content_truncated is True
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Usage propagation + aggregation
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_usage_is_propagated_from_model_response_to_result():
+    executor, _, _ = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: PASS\nREASON: ok", prompt_tokens=42, completion_tokens=7),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    usage = outcome.results[0].usage
+    assert usage.prompt_tokens == 42
+    assert usage.completion_tokens == 7
+    assert usage.total_tokens == 49
+
+
+def test_total_usage_sums_across_multiple_outputs():
+    decl_a = _file_output(name="a")
+    decl_b = _file_output(name="b")
+    executor, _, _ = _make_executor(
+        contents={
+            "produced-a": LoadedFileContent(text="pa"),
+            "produced-b": LoadedFileContent(text="pb"),
+            "bench-1": LoadedFileContent(text="b"),
+        },
+        response=_ok_response("VERDICT: PASS\nREASON: ok", prompt_tokens=10, completion_tokens=5),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[decl_a, decl_b],
+        raw_output={"a": {"file_id": "produced-a"}, "b": {"file_id": "produced-b"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    total = outcome.total_usage
+    assert total.prompt_tokens == 20
+    assert total.completion_tokens == 10
+    assert total.total_tokens == 30
+
+
+def test_outcome_failures_and_by_name_helpers():
+    decl_a = _file_output(name="a")
+    decl_b = _file_output(name="b")
+    executor, _, _ = _make_executor(
+        contents={
+            "produced-a": LoadedFileContent(text="pa"),
+            "produced-b": LoadedFileContent(text="pb"),
+            "bench-1": LoadedFileContent(text="b"),
+        },
+    )
+
+    # First call: a passes
+    executor._model_invoker = StubModelInvoker(  # type: ignore[attr-defined]
+        response=_ok_response("VERDICT: PASS\nREASON: ok")
+    )
+    outcome_a = executor.check_all(
+        declared_outputs=[decl_a],
+        raw_output={"a": {"file_id": "produced-a"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    # Second call: b fails
+    executor._model_invoker = StubModelInvoker(  # type: ignore[attr-defined]
+        response=_ok_response("VERDICT: FAIL\nREASON: missing")
+    )
+    outcome_b = executor.check_all(
+        declared_outputs=[decl_b],
+        raw_output={"b": {"file_id": "produced-b"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome_a.has_failures is False
+    assert outcome_b.has_failures is True
+    assert outcome_b.by_name()["b"].status == FileOutputCheckStatus.FAILED
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Prompt construction
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_prompt_includes_user_prompt_benchmark_and_produced_sections():
+    executor, _, invoker = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text="HELLO_PRODUCED"),
+            "bench-1": LoadedFileContent(text="HELLO_BENCHMARK"),
+        },
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+    )
+    declared = _file_output(prompt="My specific criteria.")
+    executor.check_all(
+        declared_outputs=[declared],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    prompt_text = invoker.calls[0]["prompt"]
+    assert "My specific criteria." in prompt_text
+    assert "HELLO_PRODUCED" in prompt_text
+    assert "HELLO_BENCHMARK" in prompt_text
+    assert "VERDICT: PASS" in prompt_text  # format instruction
+
+
+def test_invoker_receives_tenant_provider_model_and_settings():
+    executor, _, invoker = _make_executor(
+        contents=_ok_loader_contents(),
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+    )
+    executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {"file_id": "produced-1"}},
+        tenant_id="t-tenant",
+        model_provider="anthropic",
+        model_name="claude-sonnet",
+        model_settings={"temperature": 0.0},
+    )
+    call = invoker.calls[0]
+    assert call["tenant_id"] == "t-tenant"
+    assert call["model_provider"] == "anthropic"
+    assert call["model_name"] == "claude-sonnet"
+    assert call["model_settings"] == {"temperature": 0.0}
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Edge cases
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("alias_key", ["file_id", "upload_file_id", "tool_file_id"])
+def test_produced_file_id_alias_keys_are_recognized(alias_key: str):
+    executor, _, _ = _make_executor(
+        contents={
+            "produced-1": LoadedFileContent(text="p"),
+            "bench-1": LoadedFileContent(text="b"),
+        },
+        response=_ok_response("VERDICT: PASS\nREASON: ok"),
+    )
+    outcome = executor.check_all(
+        declared_outputs=[_file_output()],
+        raw_output={"report": {alias_key: "produced-1"}},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert outcome.results[0].status == FileOutputCheckStatus.PASSED
+
+
+def test_usage_addition_associates_currency_when_present():
+    a = FileOutputCheckUsage(
+        prompt_tokens=1, completion_tokens=2, total_tokens=3, total_price=Decimal("0.5"), currency="USD"
+    )
+    b = FileOutputCheckUsage(
+        prompt_tokens=4, completion_tokens=5, total_tokens=9, total_price=Decimal("1.5"), currency="EUR"
+    )
+    summed = a + b
+    assert summed.prompt_tokens == 5
+    assert summed.completion_tokens == 7
+    assert summed.total_tokens == 12
+    assert summed.total_price == Decimal("2.0")
+    assert summed.currency == "USD"  # a had positive price → wins
+
+
+def test_empty_check_all_returns_empty_outcome():
+    executor, _, _ = _make_executor()
+    outcome = executor.check_all(
+        declared_outputs=[],
+        raw_output={},
+        tenant_id="t-1",
+        model_provider="openai",
+        model_name="gpt-4",
+    )
+    assert isinstance(outcome, FileOutputCheckOutcome)
+    assert outcome.results == ()
+    assert outcome.has_failures is False

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_output_check_model_invoker.py

@@ -0,0 +1,183 @@
+"""Unit tests for ModelRuntimeOutputCheckInvoker (Stage 4 §6).
+
+Wraps :class:`ModelManager.invoke_llm` (``stream=False``); every provider
+exception is normalized to :class:`OutputCheckModelInvocationError` so the
+executor has a single error surface to handle.
+"""
+
+from __future__ import annotations
+
+from decimal import Decimal
+from unittest.mock import MagicMock
+
+import pytest
+
+from core.workflow.nodes.agent_v2.output_check_executor import (
+    FileOutputCheckUsage,
+    OutputCheckModelInvocationError,
+)
+from core.workflow.nodes.agent_v2.output_check_model_invoker import (
+    ModelRuntimeOutputCheckInvoker,
+)
+from graphon.model_runtime.entities.llm_entities import LLMResult, LLMUsage
+from graphon.model_runtime.entities.message_entities import (
+    AssistantPromptMessage,
+    UserPromptMessage,
+)
+from graphon.model_runtime.entities.model_entities import ModelType
+
+
+def _llm_usage(*, prompt: int = 10, completion: int = 5, latency: float = 0.5) -> LLMUsage:
+    usage = LLMUsage.empty_usage()
+    usage.prompt_tokens = prompt
+    usage.completion_tokens = completion
+    usage.total_tokens = prompt + completion
+    usage.total_price = Decimal("0.002")
+    usage.currency = "USD"
+    usage.latency = latency
+    return usage
+
+
+def _llm_result(*, content: str = "VERDICT: PASS\nREASON: ok") -> LLMResult:
+    return LLMResult(
+        id=None,
+        model="gpt-4",
+        prompt_messages=[],
+        message=AssistantPromptMessage(content=content),
+        usage=_llm_usage(),
+    )
+
+
+def _make_manager_returning(result: LLMResult) -> MagicMock:
+    instance = MagicMock()
+    instance.invoke_llm.return_value = result
+    manager = MagicMock()
+    manager.get_model_instance.return_value = instance
+    return manager
+
+
+def test_invoke_happy_path_returns_text_and_usage():
+    manager = _make_manager_returning(_llm_result(content="VERDICT: PASS\nREASON: looks good"))
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    response = invoker.invoke(
+        tenant_id="tenant-1",
+        model_provider="openai",
+        model_name="gpt-4",
+        prompt="hello",
+        model_settings={"temperature": 0.0},
+    )
+
+    assert response.text == "VERDICT: PASS\nREASON: looks good"
+    assert isinstance(response.usage, FileOutputCheckUsage)
+    assert response.usage.prompt_tokens == 10
+    assert response.usage.completion_tokens == 5
+    assert response.usage.total_tokens == 15
+    assert response.usage.latency_ms == 500  # 0.5s → 500ms
+
+
+def test_invoke_passes_args_to_model_manager_correctly():
+    manager = _make_manager_returning(_llm_result())
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    invoker.invoke(
+        tenant_id="tenant-1",
+        model_provider="anthropic",
+        model_name="claude-sonnet",
+        prompt="check this",
+        model_settings={"max_tokens": 100},
+    )
+
+    manager.get_model_instance.assert_called_once_with(
+        tenant_id="tenant-1",
+        provider="anthropic",
+        model_type=ModelType.LLM,
+        model="claude-sonnet",
+    )
+    invoke_call = manager.get_model_instance.return_value.invoke_llm
+    args, kwargs = invoke_call.call_args
+    assert kwargs["stream"] is False
+    assert kwargs["model_parameters"] == {"max_tokens": 100}
+    # Prompt message is a single UserPromptMessage with our prompt text.
+    msgs = kwargs["prompt_messages"]
+    assert len(msgs) == 1
+    assert isinstance(msgs[0], UserPromptMessage)
+    assert msgs[0].content == "check this"
+
+
+def test_invoke_with_no_settings_uses_empty_dict():
+    manager = _make_manager_returning(_llm_result())
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    invoker.invoke(
+        tenant_id="t",
+        model_provider="openai",
+        model_name="gpt-4",
+        prompt="x",
+    )
+
+    kwargs = manager.get_model_instance.return_value.invoke_llm.call_args.kwargs
+    assert kwargs["model_parameters"] == {}
+
+
+def test_provider_error_normalized_to_output_check_model_invocation_error():
+    manager = MagicMock()
+    manager.get_model_instance.side_effect = RuntimeError("provider exploded")
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    with pytest.raises(OutputCheckModelInvocationError, match="provider exploded") as exc_info:
+        invoker.invoke(
+            tenant_id="t",
+            model_provider="openai",
+            model_name="gpt-4",
+            prompt="x",
+        )
+    # Original exception is chained so logs preserve the root cause.
+    assert exc_info.value.__cause__ is not None
+
+
+def test_non_llm_result_normalized_to_invocation_error():
+    """Defensive: a provider returning a Generator while we requested
+    ``stream=False`` should not produce a cryptic AttributeError later."""
+    instance = MagicMock()
+    instance.invoke_llm.return_value = iter(["chunk"])  # generator-shaped, not LLMResult
+    manager = MagicMock()
+    manager.get_model_instance.return_value = instance
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    with pytest.raises(OutputCheckModelInvocationError, match="LLMResult"):
+        invoker.invoke(
+            tenant_id="t",
+            model_provider="openai",
+            model_name="gpt-4",
+            prompt="x",
+        )
+
+
+def test_assistant_content_list_is_flattened_to_string():
+    """Multimodal content lists are flattened to one string via ``data``/``text``
+    attributes; defensive in case a provider returns content parts.
+
+    Constructed via ``model_construct`` to bypass Pydantic's strict content
+    validator — we want to verify the invoker's tolerance, not whatever
+    AssistantPromptMessage's accepted-shape happens to be today.
+    """
+
+    class _TextPart:
+        def __init__(self, text: str) -> None:
+            self.text = text
+
+    msg = AssistantPromptMessage.model_construct(content=[_TextPart("VERDICT: PASS"), _TextPart("REASON: ok")])
+    result = _llm_result()
+    result.message = msg
+    manager = _make_manager_returning(result)
+    invoker = ModelRuntimeOutputCheckInvoker(model_manager_factory=lambda _: manager)
+
+    response = invoker.invoke(
+        tenant_id="t",
+        model_provider="openai",
+        model_name="gpt-4",
+        prompt="x",
+    )
+    assert "VERDICT: PASS" in response.text
+    assert "REASON: ok" in response.text

api/tests/unit_tests/core/workflow/nodes/agent_v2/test_upload_file_content_loader.py

@@ -0,0 +1,165 @@
+"""Defensive tests for UploadFileContentLoader (Stage 4 §6).
+
+Loader must never raise on pathological inputs and must map all failure
+modes to a deterministic ``LoadedFileContent`` / ``None`` shape so the
+executor can produce a clean SKIPPED result.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from sqlalchemy.exc import DataError, SQLAlchemyError
+
+from core.workflow.nodes.agent_v2.output_check_executor import LoadedFileContent
+from core.workflow.nodes.agent_v2.upload_file_content_loader import UploadFileContentLoader
+
+VALID_UUID = "550e8400-e29b-41d4-a716-446655440000"
+
+
+def _make_upload_file(*, tenant_id: str = "tenant-1", extension: str = "pdf"):
+    """Tiny stand-in for ``UploadFile`` ORM object."""
+    upload = MagicMock()
+    upload.tenant_id = tenant_id
+    upload.extension = extension
+    return upload
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Pre-DB short-circuits
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_empty_inputs_short_circuit_without_db_hit():
+    loader = UploadFileContentLoader()
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        assert loader.load(file_id="", tenant_id="t-1") is None
+        assert loader.load(file_id="abc", tenant_id="") is None
+    factory.create_session.assert_not_called()
+
+
+@pytest.mark.parametrize("bad", ["not-a-uuid", "🤖", "../etc/passwd", VALID_UUID + "-trailing"])
+def test_non_uuid_file_ids_short_circuit_without_db_hit(bad: str):
+    loader = UploadFileContentLoader()
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        assert loader.load(file_id=bad, tenant_id="t-1") is None
+    factory.create_session.assert_not_called()
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# DB errors / cross-tenant
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize("error_cls", [SQLAlchemyError, DataError])
+def test_db_errors_return_none(error_cls):
+    loader = UploadFileContentLoader()
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        factory.create_session.return_value.__enter__.return_value.scalar.side_effect = (
+            error_cls("boom") if error_cls is SQLAlchemyError else error_cls("x", "y", "z")
+        )
+        assert loader.load(file_id=VALID_UUID, tenant_id="t-1") is None
+
+
+def test_missing_upload_file_returns_none():
+    loader = UploadFileContentLoader()
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = None
+        assert loader.load(file_id=VALID_UUID, tenant_id="t-1") is None
+
+
+def test_cross_tenant_returns_none():
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(tenant_id="other-tenant")
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        assert loader.load(file_id=VALID_UUID, tenant_id="t-1") is None
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Extension routing
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+@pytest.mark.parametrize(
+    "extension",
+    ["png", "jpg", "mp4", "zip", "exe", ".gif", "WEBP"],  # mix lowercase / dotted / upper
+)
+def test_unsupported_extensions_short_circuit_before_extraction(extension: str):
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(extension=extension)
+    with (
+        patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory,
+        patch(
+            "core.workflow.nodes.agent_v2.upload_file_content_loader.ExtractProcessor.load_from_upload_file"
+        ) as extract,
+    ):
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        result = loader.load(file_id=VALID_UUID, tenant_id="tenant-1")
+    assert result == LoadedFileContent(text="", unsupported=True)
+    extract.assert_not_called()
+
+
+def test_extension_with_leading_dot_is_normalized():
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(extension=".png")
+    with patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory:
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        result = loader.load(file_id=VALID_UUID, tenant_id="tenant-1")
+    assert result is not None
+    assert result.unsupported is True
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Extraction success / failure
+# ──────────────────────────────────────────────────────────────────────────────
+
+
+def test_successful_extraction_returns_loaded_content():
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(extension="pdf")
+    with (
+        patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory,
+        patch(
+            "core.workflow.nodes.agent_v2.upload_file_content_loader.ExtractProcessor.load_from_upload_file"
+        ) as extract,
+    ):
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        extract.return_value = "extracted pdf text"
+        result = loader.load(file_id=VALID_UUID, tenant_id="tenant-1")
+    assert result == LoadedFileContent(text="extracted pdf text")
+    extract.assert_called_once_with(upload, return_text=True)
+
+
+def test_extractor_raising_falls_back_to_unsupported():
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(extension="pdf")
+    with (
+        patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory,
+        patch(
+            "core.workflow.nodes.agent_v2.upload_file_content_loader.ExtractProcessor.load_from_upload_file"
+        ) as extract,
+    ):
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        extract.side_effect = RuntimeError("corrupt file")
+        result = loader.load(file_id=VALID_UUID, tenant_id="tenant-1")
+    assert result == LoadedFileContent(text="", unsupported=True)
+
+
+def test_extractor_returning_non_string_is_treated_as_unsupported():
+    """``ExtractProcessor.load_from_upload_file`` returns ``list[Document]`` when
+    ``return_text=False``; defensive path in case any caller refactors away the
+    flag."""
+    loader = UploadFileContentLoader()
+    upload = _make_upload_file(extension="pdf")
+    with (
+        patch("core.workflow.nodes.agent_v2.upload_file_content_loader.session_factory") as factory,
+        patch(
+            "core.workflow.nodes.agent_v2.upload_file_content_loader.ExtractProcessor.load_from_upload_file"
+        ) as extract,
+    ):
+        factory.create_session.return_value.__enter__.return_value.scalar.return_value = upload
+        extract.return_value = [object()]  # not a string
+        result = loader.load(file_id=VALID_UUID, tenant_id="tenant-1")
+    assert result == LoadedFileContent(text="", unsupported=True)