diff --git a/docker-compose.yml b/docker-compose.yml
index 378c01bbbf..05be0bf15d 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -41,6 +41,8 @@ services:
       - BACKEND_PORT=8000
       - XAGENT_UPLOADS_DIR=/root/.xagent/uploads
       - XAGENT_MAX_UPLOAD_SIZE=${XAGENT_MAX_UPLOAD_SIZE:-100M}
+      - XAGENT_ARTIFACT_VALIDATION_MAX_BYTES=${XAGENT_ARTIFACT_VALIDATION_MAX_BYTES:-32M}
+      - XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS=${XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS:-8}
       - XAGENT_REDIS_URL=redis://redis:6379/0
       - XAGENT_HOT_PATH_CACHE_ENABLED=${XAGENT_HOT_PATH_CACHE_ENABLED:-true}
       - XAGENT_CELERY_ENABLED=${XAGENT_CELERY_ENABLED:-true}
@@ -92,6 +94,8 @@ services:
       - POSTGRES_PASSWORD=${POSTGRES_PASSWORD:-xagent_password}
       - XAGENT_UPLOADS_DIR=/root/.xagent/uploads
       - XAGENT_MAX_UPLOAD_SIZE=${XAGENT_MAX_UPLOAD_SIZE:-100M}
+      - XAGENT_ARTIFACT_VALIDATION_MAX_BYTES=${XAGENT_ARTIFACT_VALIDATION_MAX_BYTES:-32M}
+      - XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS=${XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS:-8}
       - XAGENT_REDIS_URL=redis://redis:6379/0
       - XAGENT_HOT_PATH_CACHE_ENABLED=${XAGENT_HOT_PATH_CACHE_ENABLED:-true}
       - XAGENT_CELERY_ENABLED=${XAGENT_CELERY_ENABLED:-true}
diff --git a/example.env b/example.env
index 2a48e7a8a7..345fb256d0 100644
--- a/example.env
+++ b/example.env
@@ -867,6 +867,11 @@ XAGENT_EXTERNAL_SKILLS_LIBRARY_DIRS=""
 # Supports raw bytes or human-readable values like 100M, 1G, 512K.
 XAGENT_MAX_UPLOAD_SIZE="100M"
 
+# Format-readability validation for generated artifacts. Unsupported formats,
+# absent optional parsers and exhausted budgets are reported as unchecked.
+XAGENT_ARTIFACT_VALIDATION_MAX_BYTES=32M
+XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS=8
+
 # Durable file storage for user-visible uploads and registered workspace outputs.
 # Defaults to file://$XAGENT_STORAGE_ROOT/files for local development.
 # For S3-compatible storage, use a URI with bucket and optional prefix.
diff --git a/frontend/src/components/file/artifact-validation.test.tsx b/frontend/src/components/file/artifact-validation.test.tsx
new file mode 100644
index 0000000000..280d489425
--- /dev/null
+++ b/frontend/src/components/file/artifact-validation.test.tsx
@@ -0,0 +1,116 @@
+/// <reference types="@testing-library/jest-dom/vitest" />
+import React from 'react'
+import { act, cleanup, fireEvent, render, screen, waitFor } from '@testing-library/react'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { ArtifactValidation } from './artifact-validation'
+import { FileAccessProvider, defaultFileAccessPolicy, createPublicFileAccessPolicy } from '@/contexts/file-access-context'
+import { I18nProvider } from '@/contexts/i18n-context'
+import { InlineFilePreview } from './inline-file-preview'
+
+const reportHeaders = { 'content-type': 'application/vnd.xagent.validation+json' }
+const report = (status: string, message = '', supported = true) => new Response(JSON.stringify({ status, supported, sha256: '0'.repeat(64), checks: [{ status, message }] }), { headers: reportHeaders })
+const wrap = (children: React.ReactNode, request = vi.fn(), extra = {}) => <I18nProvider>
+  <FileAccessProvider policy={{ ...defaultFileAccessPolicy, request, ...extra }}>{children}</FileAccessProvider>
+</I18nProvider>
+
+afterEach(() => { cleanup(); vi.unstubAllGlobals(); vi.useRealTimers(); localStorage.removeItem('app_locale') })
+
+describe('ArtifactValidation', () => {
+  it.each(['valid', 'invalid', 'unchecked'] as const)('shows %s without hiding repair/download access', async status => {
+    const request = vi.fn().mockResolvedValue(report(status, status === 'invalid' ? 'Corrupt package' : ''))
+    const { container } = render(wrap(<ArtifactValidation fileId="one"><a href="/download">artifact</a></ArtifactValidation>, request))
+    expect(container.querySelector('[data-artifact-validation="checking"]')).toBeTruthy()
+    await waitFor(() => expect(container.querySelector(`[data-artifact-validation="${status}"]`)).toBeTruthy())
+    expect(screen.getByRole('link', { name: 'artifact' })).toHaveAttribute('href', '/download')
+    expect(request).toHaveBeenCalledWith('/api/files/preview/one?validation_only=true', expect.objectContaining({ cache: 'no-store', signal: expect.any(AbortSignal) }))
+  })
+
+  it('rechecks the same file id after repair without reusing a cached pass', async () => {
+    const request = vi.fn().mockResolvedValueOnce(report('valid')).mockResolvedValueOnce(report('invalid', 'broken'))
+    const { container } = render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, request))
+    await screen.findByText('Format readable · content not verified')
+    fireEvent.click(screen.getByRole('button', { name: 'Recheck' }))
+    expect(container.querySelector('[data-artifact-validation="checking"]')).toBeTruthy()
+    await screen.findByText('File validation failed · repair required')
+    expect(request).toHaveBeenCalledTimes(2)
+  })
+
+  it.each([
+    new Response('', { status: 403 }), new Response('', { status: 500 }), report('made-up'), new Response('not json'),
+    new Response('{"status":"valid"}', { headers: reportHeaders }),
+    new Response('{"status":"valid","checks":[{"status":"valid"}]}', { headers: reportHeaders }),
+    new Response(JSON.stringify({ status: 'valid', sha256: '0'.repeat(64), checks: [{ status: 'unchecked' }] }), { headers: reportHeaders }),
+    new Response(JSON.stringify({ status: 'valid', sha256: '0'.repeat(64), checks: [{ status: 'valid' }] }), { headers: { 'content-type': 'application/json' } }),
+  ])('does not treat a failed/malformed request as a pass', async response => {
+    render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, vi.fn().mockResolvedValue(response)))
+    await screen.findByText('Unable to request file validation · try again')
+    expect(screen.queryByText('File not checked')).toBeNull()
+  })
+
+  it('ignores an old response after the file changes', async () => {
+    let finish!: (response: Response) => void
+    const request = vi.fn().mockImplementationOnce(() => new Promise(resolve => { finish = resolve })).mockResolvedValueOnce(report('invalid'))
+    const policy = { ...defaultFileAccessPolicy, request }
+    const view = (fileId: string) => <I18nProvider><FileAccessProvider policy={policy}><ArtifactValidation fileId={fileId}>file</ArtifactValidation></FileAccessProvider></I18nProvider>
+    const { rerender } = render(view('one'))
+    rerender(view('two'))
+    await screen.findByText('File validation failed · repair required')
+    await act(async () => { finish(report('valid')) })
+    expect(screen.queryByText('Format readable · content not verified')).toBeNull()
+    expect(request.mock.calls[0][1].signal.aborted).toBe(true)
+  })
+
+  it('does not request validation for policies without the capability', () => {
+    const request = vi.fn()
+    render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, request, { validationUrl: undefined }))
+    expect(request).not.toHaveBeenCalled()
+    expect(screen.queryByRole('status')).toBeNull()
+  })
+
+  it('uses the public scoped token and strips ambient authorization', async () => {
+    const fetchMock = vi.fn().mockResolvedValue(report('valid'))
+    vi.stubGlobal('fetch', fetchMock)
+    const policy = createPublicFileAccessPolicy('guest-token')
+    render(<I18nProvider><FileAccessProvider policy={policy}><ArtifactValidation fileId="one">file</ArtifactValidation></FileAccessProvider></I18nProvider>)
+    await screen.findByText('Format readable · content not verified')
+    expect(fetchMock.mock.calls[0][0]).toContain('token=guest-token&validation_only=true')
+    expect(fetchMock.mock.calls[0][1].headers.has('Authorization')).toBe(false)
+  })
+
+  it('hides unsupported-format controls using the server capability, retaining the file', async () => {
+    const request = vi.fn().mockResolvedValue(report('unchecked', 'No validator is installed for this format.', false))
+    render(wrap(<InlineFilePreview source={{ fileId: '123e4567-e89b-12d3-a456-426614174000', filename: 'notes.txt' }} />, request))
+    await waitFor(() => expect(screen.queryByRole('status')).toBeNull())
+    expect(request).toHaveBeenCalledTimes(1)
+    expect(screen.queryByRole('button', { name: 'Recheck' })).toBeNull()
+    expect(screen.getByRole('link')).toHaveTextContent('notes.txt')
+  })
+
+  it('distinguishes a network error and permits a successful retry', async () => {
+    const request = vi.fn().mockRejectedValueOnce(new TypeError('offline')).mockResolvedValueOnce(report('valid'))
+    render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, request))
+    await screen.findByText('Unable to request file validation · try again')
+    fireEvent.click(screen.getByRole('button', { name: 'Recheck' }))
+    await screen.findByText('Format readable · content not verified')
+  })
+
+  it('reports a client timeout as a request error, not a completed unchecked report', async () => {
+    vi.useFakeTimers()
+    const request = vi.fn().mockImplementation((_url, options) => new Promise((_resolve, reject) => {
+      options.signal.addEventListener('abort', () => reject(new DOMException('Aborted', 'AbortError')))
+    }))
+    render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, request))
+    await act(async () => { await vi.advanceTimersByTimeAsync(20_001) })
+    expect(screen.getByText('Unable to request file validation · try again')).toBeInTheDocument()
+    expect(request.mock.calls[0][1].signal.aborted).toBe(true)
+  })
+
+  it('uses translated machine-status labels instead of raw English parser diagnostics', async () => {
+    localStorage.setItem('app_locale', 'zh')
+    const request = vi.fn().mockResolvedValue(report('invalid', 'PDF header is missing.'))
+    render(wrap(<ArtifactValidation fileId="one">file</ArtifactValidation>, request))
+    await screen.findByText('文件校验失败 · 需要修复')
+    expect(screen.queryByText('PDF header is missing.')).toBeNull()
+    expect(screen.getByRole('button', { name: '重新检查' })).toBeInTheDocument()
+  })
+})
diff --git a/frontend/src/components/file/artifact-validation.tsx b/frontend/src/components/file/artifact-validation.tsx
new file mode 100644
index 0000000000..80d19c0e9e
--- /dev/null
+++ b/frontend/src/components/file/artifact-validation.tsx
@@ -0,0 +1,91 @@
+import React, { useEffect, useState } from 'react'
+import { useFileAccess } from '@/contexts/file-access-context'
+import { useI18n } from '@/contexts/i18n-context'
+
+type Status = 'valid' | 'invalid' | 'unchecked'
+type DisplayReport = { status: Status | 'error'; supported: boolean }
+const STATUSES: ReadonlySet<string> = new Set(['valid', 'invalid', 'unchecked'])
+
+async function readReport(response: Response): Promise<DisplayReport> {
+  if (!response.ok) throw new Error('Validation unavailable')
+  // An older backend may ignore validation_only and return the attachment
+  // itself. Never interpret arbitrary JSON file contents as a report.
+  if (response.headers.get('content-type')?.split(';')[0] !== 'application/vnd.xagent.validation+json') {
+    throw new Error('Not a validation response')
+  }
+  const data = await response.json()
+  if (!data || !STATUSES.has(data.status) ||
+      !Array.isArray(data.checks) || !data.checks.length ||
+      !data.checks.every((c: { status?: unknown } | null) => c && STATUSES.has(String(c.status)))) {
+    throw new Error('Invalid report')
+  }
+  if (data.status !== 'unchecked' &&
+      (typeof data.sha256 !== 'string' || !/^[a-f0-9]{64}$/.test(data.sha256))) {
+    throw new Error('Missing snapshot')
+  }
+  if (data.status === 'valid' && !data.checks.every((c: { status: string }) => c.status === 'valid')) {
+    throw new Error('Incomplete checks')
+  }
+  // The backend owns format support. Do not duplicate a suffix allowlist in
+  // the UI or show an endless Recheck action for formats without a validator.
+  // Machine statuses drive localized presentation; English parser diagnostics
+  // remain in the API/model report rather than leaking into the user's locale.
+  return { status: data.status, supported: data.supported !== false }
+}
+
+/** Server-authoritative, current-byte checks, independent of preview renderers. */
+export function ArtifactValidation({ fileId, children }: {
+  fileId: string
+  children: React.ReactNode
+}) {
+  const policy = useFileAccess()
+  const { t } = useI18n()
+  const [attempt, setAttempt] = useState(0)
+  const [result, setResult] = useState<DisplayReport & { key: string }>()
+  const url = policy.validationUrl?.(fileId)
+  const key = `${url}:${attempt}`
+
+  useEffect(() => {
+    if (!url) return
+    let active = true
+    const controller = new AbortController()
+    const timeout = setTimeout(() => controller.abort(), 20_000)
+    const check = async () => {
+      try {
+        const response = await policy.request(url, { signal: controller.signal, cache: 'no-store' })
+        const report = await readReport(response)
+        if (active) setResult({ key, ...report })
+      } catch {
+        if (active) setResult({ key, status: 'error', supported: true })
+      } finally {
+        clearTimeout(timeout)
+      }
+    }
+    void check()
+    return () => {
+      active = false
+      clearTimeout(timeout)
+      controller.abort()
+    }
+  }, [key, url, policy])
+
+  if (!url) return <>{children}</>
+  const current = result?.key === key ? result : undefined
+  if (current?.supported === false) return <>{children}</>
+  const label = current?.status ?? 'checking'
+  return (
+    <div data-artifact-validation={label}>
+      <div className="flex items-center gap-2 py-1 text-xs text-muted-foreground" role="status">
+        <span className={label === 'invalid' ? 'text-destructive' : undefined}>
+          {t(`files.validation.${label}`)}
+        </span>
+        {current ? (
+          <button type="button" className="underline" onClick={() => setAttempt(n => n + 1)}>
+            {t('files.validation.recheck')}
+          </button>
+        ) : null}
+      </div>
+      <React.Fragment key={key}>{children}</React.Fragment>
+    </div>
+  )
+}
diff --git a/frontend/src/components/file/inline-file-preview.test.tsx b/frontend/src/components/file/inline-file-preview.test.tsx
index 7c5ec6d611..5ea8f7e5e2 100644
--- a/frontend/src/components/file/inline-file-preview.test.tsx
+++ b/frontend/src/components/file/inline-file-preview.test.tsx
@@ -6,6 +6,12 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 const apiRequestMock = vi.hoisted(() => vi.fn())
 const toastErrorMock = vi.hoisted(() => vi.fn())
 
+// Validation has its own server-boundary/integration suite; these tests isolate
+// the existing renderer and streaming contracts from that separate request.
+vi.mock('@/components/file/artifact-validation', () => ({
+  ArtifactValidation: ({ children }: { children: React.ReactNode }) => <>{children}</>,
+}))
+
 vi.mock('@/components/ui/sonner', () => ({
   toast: { error: toastErrorMock },
 }))
diff --git a/frontend/src/components/file/inline-file-preview.tsx b/frontend/src/components/file/inline-file-preview.tsx
index 3529a702cf..5799e6e57d 100644
--- a/frontend/src/components/file/inline-file-preview.tsx
+++ b/frontend/src/components/file/inline-file-preview.tsx
@@ -4,6 +4,7 @@ import { FileText, Loader2, Video, Volume2 } from 'lucide-react'
 import { DocxPreviewRenderer } from '@/components/file/docx-preview-renderer'
 import { ExcelPreviewRenderer } from '@/components/file/excel-preview-renderer'
 import { PptxPreviewRenderer } from '@/components/file/pptx-preview-renderer'
+import { ArtifactValidation } from '@/components/file/artifact-validation'
 import { toast } from '@/components/ui/sonner'
 import { cn, getApiUrl } from '@/lib/utils'
 import { useFileAccess, type FileAccessPolicy } from '@/contexts/file-access-context'
@@ -865,7 +866,19 @@ function ExternalPreviewPlaceholder({
   )
 }
 
-export function InlineFilePreview({
+export function InlineFilePreview(props: InlineFilePreviewProps) {
+  const fileAccess = useFileAccess()
+  const fileId = props.source.fileId && resolveInlineFileId(props.source.fileId)
+  const kind = getInlineFilePreviewKind(props.source)
+  const content = <InlineFilePreviewContent {...props} />
+  // External URLs never enter the authenticated validation boundary. Audio and
+  // video keep progressive delivery; their decoders are not covered yet.
+  if (!fileId || !fileAccess.validationUrl || kind === 'audio' || kind === 'video' ||
+      !getPreviewUrlTrust({ ...props.source, fileId }, getApiUrl()).isTrusted) return content
+  return <ArtifactValidation key={fileId} fileId={fileId}>{content}</ArtifactValidation>
+}
+
+function InlineFilePreviewContent({
   source,
   className,
   imageClassName,
diff --git a/frontend/src/contexts/file-access-context.tsx b/frontend/src/contexts/file-access-context.tsx
index 94951bd4ec..d88134b82f 100644
--- a/frontend/src/contexts/file-access-context.tsx
+++ b/frontend/src/contexts/file-access-context.tsx
@@ -17,6 +17,7 @@ export interface FileAccessPolicy {
   inlineDownloadUrl: (fileId: string) => string
   relativePreviewUrl: (fileId: string, relativePath: string) => string
   pdfPreviewUrl?: (fileId: string) => string
+  validationUrl?: (fileId: string) => string
   /**
    * Execute under this policy's credential boundary. The built-in default
    * attaches Bearer authorization. The public policy forces same-origin
@@ -77,6 +78,7 @@ export const defaultFileAccessPolicy: FileAccessPolicy = {
     return getApiUrl() ? url.toString() : `${url.pathname}${url.search}${url.hash}`
   },
   pdfPreviewUrl: (fileId) => buildUrl(`${FILES_API_PREFIX}/preview-pdf/${encodeFileId(fileId)}`),
+  validationUrl: (fileId) => buildUrl(`${FILES_API_PREFIX}/preview/${encodeFileId(fileId)}?validation_only=true`),
   request: apiRequest,
   listFiles: (query) => {
     const params = new URLSearchParams({ page: "1", size: "20" })
@@ -197,6 +199,7 @@ export function createPublicFileAccessPolicy(accessToken: string): FileAccessPol
     downloadUrl: inlineDownloadUrl,
     inlinePreviewUrl,
     inlineDownloadUrl,
+    validationUrl: (fileId) => `${inlinePreviewUrl(fileId)}&validation_only=true`,
     relativePreviewUrl: (fileId, relativePath) => {
       const url = new URL(
         inlinePreviewUrl(fileId),
diff --git a/frontend/src/i18n/locales/en.ts b/frontend/src/i18n/locales/en.ts
index 049347c754..b75c6c3c86 100644
--- a/frontend/src/i18n/locales/en.ts
+++ b/frontend/src/i18n/locales/en.ts
@@ -1497,6 +1497,14 @@ Build when you need.`,
       page: "Page {page} of {pages}",
       next: "Next",
     },
+    validation: {
+      checking: "Checking file format…",
+      valid: "Format readable · content not verified",
+      invalid: "File validation failed · repair required",
+      unchecked: "File not checked",
+      error: "Unable to request file validation · try again",
+      recheck: "Recheck",
+    },
     previewDialog: {
       buttons: {
         download: "Download",
diff --git a/frontend/src/i18n/locales/zh.ts b/frontend/src/i18n/locales/zh.ts
index 1c16aaff9a..f0ea6336d5 100644
--- a/frontend/src/i18n/locales/zh.ts
+++ b/frontend/src/i18n/locales/zh.ts
@@ -1497,6 +1497,14 @@ const zh = {
       page: "第 {page} 页，共 {pages} 页",
       next: "下一页",
     },
+    validation: {
+      checking: "正在检查文件格式…",
+      valid: "格式可读 · 内容尚未核验",
+      invalid: "文件校验失败 · 需要修复",
+      unchecked: "文件未校验",
+      error: "无法请求文件校验 · 请重试",
+      recheck: "重新检查",
+    },
     previewDialog: {
       buttons: {
         download: "下载",
diff --git a/frontend/vitest.widget.config.ts b/frontend/vitest.widget.config.ts
index 4f62f648b3..ada016b0e9 100644
--- a/frontend/vitest.widget.config.ts
+++ b/frontend/vitest.widget.config.ts
@@ -15,6 +15,7 @@ const widgetConfig = mergeConfig(baseConfig, defineConfig({
         "src/components/file/file-preview-content.tsx",
         "src/components/file/file-viewer.tsx",
         "src/components/file/inline-file-preview.tsx",
+        "src/components/file/artifact-validation.tsx",
         "src/components/file/pptx-preview-renderer.tsx",
         "src/components/task/task-conversation-panel.tsx",
         "src/components/ui/markdown-renderer.tsx",
@@ -116,6 +117,9 @@ const widgetConfig = mergeConfig(baseConfig, defineConfig({
         "src/components/file/inline-file-preview.tsx": {
           statements: 70, branches: 55, functions: 60, lines: 70,
         },
+        "src/components/file/artifact-validation.tsx": {
+          statements: 90, branches: 80, functions: 90, lines: 90,
+        },
         "src/components/file/pptx-preview-renderer.tsx": {
           statements: 45, branches: 35, functions: 30, lines: 45,
         },
@@ -148,6 +152,7 @@ export default defineConfig({
       "src/components/file/file-preview-content.test.tsx",
       "src/components/file/file-viewer.test.tsx",
       "src/components/file/inline-file-preview.test.tsx",
+      "src/components/file/artifact-validation.test.tsx",
       "src/components/file/pptx-preview-renderer.test.tsx",
       "src/components/layout/sidebar.test.tsx",
       "src/components/pages/login.test.tsx",
diff --git a/pyproject.toml b/pyproject.toml
index 901fbde9b4..3ee8e8af83 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -47,6 +47,7 @@ dependencies = [
   "celery[redis]>=5.4.0,<6.0.0",
   "redis>=5.0.0",
   "pillow >= 10.0.0",
+  "defusedxml>=0.7.1",
   "pypinyin>=0.53.0",
   "cairosvg >= 2.7.1",
   "lancedb>=0.24.2",
@@ -298,9 +299,13 @@ module = [
   "asyncssh",
   "bashlex",
   "docx",
+  "defusedxml",
+  "defusedxml.*",
   "pptx",
   "pptx.*",
   "pdfplumber",
+  "pypdf",
+  "pypdf.*",
   "unstructured.*",
   "fitz",
   "pandas",
diff --git a/src/xagent/config.py b/src/xagent/config.py
index bcea7ac5a9..81453ff3d8 100644
--- a/src/xagent/config.py
+++ b/src/xagent/config.py
@@ -1811,6 +1811,29 @@ def get_frontend_dist_dir() -> Path:
     return get_web_dir() / "frontend_dist"
 
 
+ARTIFACT_VALIDATION_MAX_BYTES = "XAGENT_ARTIFACT_VALIDATION_MAX_BYTES"
+ARTIFACT_VALIDATION_TIMEOUT_SECONDS = "XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS"
+
+
+def get_artifact_validation_max_bytes() -> int:
+    """Maximum snapshot bytes to format-check (larger files remain unchecked)."""
+    return _parse_size_bytes(
+        os.getenv(ARTIFACT_VALIDATION_MAX_BYTES) or "32M", ARTIFACT_VALIDATION_MAX_BYTES
+    )
+
+
+def get_artifact_validation_timeout_seconds() -> float:
+    """Hard timeout for each isolated artifact parser process."""
+    import math
+
+    value = float(os.getenv(ARTIFACT_VALIDATION_TIMEOUT_SECONDS) or "8")
+    if not math.isfinite(value) or value <= 0:
+        raise ValueError(
+            f"{ARTIFACT_VALIDATION_TIMEOUT_SECONDS} must be positive and finite"
+        )
+    return value
+
+
 def get_max_upload_size_bytes() -> int:
     """Get the maximum allowed upload size in bytes.
 
@@ -1833,9 +1856,14 @@ def get_max_upload_size_bytes() -> int:
     if not env_value:
         return 100 * 1024 * 1024
 
-    normalized = env_value.strip().upper()
-    if not normalized:
+    if not env_value.strip():
         return 100 * 1024 * 1024
+    return _parse_size_bytes(env_value, MAX_UPLOAD_SIZE)
+
+
+def _parse_size_bytes(env_value: str, setting: str) -> int:
+    """Shared positive byte-size parser for upload and validation budgets."""
+    normalized = env_value.strip().upper()
 
     suffix_multipliers = [
         ("GB", 1024 * 1024 * 1024),
@@ -1853,27 +1881,23 @@ def get_max_upload_size_bytes() -> int:
             number_part = normalized[: -len(suffix)].strip()
             if not number_part:
                 raise ValueError(
-                    f"Invalid {MAX_UPLOAD_SIZE} value: {env_value!r}. Missing numeric value."
+                    f"Invalid {setting} value: {env_value!r}. Missing numeric value."
                 )
             try:
                 result = int(float(number_part) * multiplier)
-            except ValueError as exc:
-                raise ValueError(
-                    f"Invalid {MAX_UPLOAD_SIZE} value: {env_value!r}."
-                ) from exc
+            except (ValueError, OverflowError) as exc:
+                raise ValueError(f"Invalid {setting} value: {env_value!r}.") from exc
             break
 
     if result is None:
         try:
             result = int(float(normalized))
-        except ValueError as exc:
-            raise ValueError(
-                f"Invalid {MAX_UPLOAD_SIZE} value: {env_value!r}."
-            ) from exc
+        except (ValueError, OverflowError) as exc:
+            raise ValueError(f"Invalid {setting} value: {env_value!r}.") from exc
 
     if result <= 0:
         raise ValueError(
-            f"Invalid {MAX_UPLOAD_SIZE} value: {env_value!r}. Value must be positive."
+            f"Invalid {setting} value: {env_value!r}. Value must be positive."
         )
 
     return result
diff --git a/src/xagent/core/artifact_validation/__init__.py b/src/xagent/core/artifact_validation/__init__.py
new file mode 100644
index 0000000000..dc9a3ae392
--- /dev/null
+++ b/src/xagent/core/artifact_validation/__init__.py
@@ -0,0 +1,12 @@
+"""Modular artifact readability checks, independent of skills and agent patterns."""
+
+from .models import ArtifactCheck, ArtifactContent, ValidationLimits, ValidationReport
+from .registry import ArtifactCheckRegistry
+
+__all__ = [
+    "ArtifactCheck",
+    "ArtifactCheckRegistry",
+    "ArtifactContent",
+    "ValidationLimits",
+    "ValidationReport",
+]
diff --git a/src/xagent/core/artifact_validation/defaults.py b/src/xagent/core/artifact_validation/defaults.py
new file mode 100644
index 0000000000..aa582cafba
--- /dev/null
+++ b/src/xagent/core/artifact_validation/defaults.py
@@ -0,0 +1,26 @@
+"""Composition root. Add checks here without modifying tools or delivery code."""
+
+from .formats import check_csv, check_image, check_pdf
+from .models import ArtifactCheck
+from .office import check_office_document, check_package
+from .registry import ArtifactCheckRegistry
+
+
+def default_registry() -> ArtifactCheckRegistry:
+    registry = ArtifactCheckRegistry()
+    office = frozenset({".xlsx", ".docx", ".pptx"})
+    for check in (
+        ArtifactCheck("office-package", office, check_package),
+        ArtifactCheck("office-reader", office, check_office_document),
+        ArtifactCheck("csv-reader", frozenset({".csv", ".tsv"}), check_csv),
+        ArtifactCheck("pdf-reader", frozenset({".pdf"}), check_pdf),
+        ArtifactCheck(
+            "image-decoder",
+            frozenset(
+                {".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tif", ".tiff"}
+            ),
+            check_image,
+        ),
+    ):
+        registry.register(check)
+    return registry
diff --git a/src/xagent/core/artifact_validation/formats.py b/src/xagent/core/artifact_validation/formats.py
new file mode 100644
index 0000000000..cf88126e80
--- /dev/null
+++ b/src/xagent/core/artifact_validation/formats.py
@@ -0,0 +1,142 @@
+"""Independent format checks; no business-content or minimum-size heuristics."""
+
+import csv
+import zlib
+from io import BytesIO, StringIO
+
+from .models import ArtifactContent, InvalidArtifact, UncheckedArtifact
+
+
+def check_csv(content: ArtifactContent) -> None:
+    data = content.data
+    try:
+        # BOMs are explicit encoding declarations. Do not guess latin-1 and
+        # thereby accept arbitrary binary files as a one-column CSV.
+        encoding = (
+            "utf-16" if data.startswith((b"\xff\xfe", b"\xfe\xff")) else "utf-8-sig"
+        )
+        text = data.decode(encoding)
+    except UnicodeError as exc:
+        raise UncheckedArtifact(
+            "CSV encoding is not UTF-8 or BOM-declared UTF-16."
+        ) from exc
+    if "\x00" in text:
+        raise InvalidArtifact("CSV contains binary NUL characters.")
+    try:
+        dialect = csv.Sniffer().sniff(text[:8192], delimiters=",;\t|")
+    except csv.Error:
+        dialect = csv.excel
+    try:
+        csv.field_size_limit(content.limits.max_bytes)
+        for index, _row in enumerate(
+            csv.reader(StringIO(text, newline=""), dialect, strict=True)
+        ):
+            if index >= content.limits.max_units:
+                raise UncheckedArtifact("CSV exceeds the row budget.")
+        # Ragged rows, an empty file, or a single column can all be legitimate.
+        # Schema/business expectations belong to additional explicit checks.
+    except csv.Error as exc:
+        # Lenient readers may recover nonstandard quoting, but can also merge
+        # truncated records silently. Neither corruption nor validity is proven.
+        raise UncheckedArtifact(
+            "CSV quoting or record structure is ambiguous."
+        ) from exc
+
+
+def check_pdf(content: ArtifactContent) -> None:
+    from pypdf import PdfReader
+    from pypdf.errors import DependencyError, PdfReadError, PyPdfError
+    from pypdf.generic import ArrayObject, EncodedStreamObject
+
+    if not content.data.lstrip().startswith(b"%PDF-"):
+        raise InvalidArtifact("PDF header is missing.")
+    try:
+        # Validate readability, not strict PDF conformance. Real readers can
+        # recover common producer defects such as an inaccurate xref offset.
+        reader = PdfReader(BytesIO(content.data), strict=False)
+        if reader.is_encrypted:
+            raise UncheckedArtifact("Encrypted PDFs require a password to validate.")
+        # A fixed parser ceiling, not partial validation: larger PDFs are
+        # unchecked in their entirety, never certified from a page sample.
+        page_limit = min(content.limits.max_units, 500)
+        if len(reader.pages) > page_limit:
+            raise UncheckedArtifact(
+                f"PDF exceeds the page budget ({page_limit} pages)."
+            )
+        expanded = 0
+        for page in reader.pages:
+            raw_contents = page.get("/Contents")
+            if raw_contents is not None:
+                raw_contents = raw_contents.get_object()
+                parts = (
+                    raw_contents
+                    if isinstance(raw_contents, ArrayObject)
+                    else [raw_contents]
+                )
+                for part in parts:
+                    part = part.get_object()
+                    if not isinstance(part, EncodedStreamObject):
+                        continue
+                    filters = part.get("/Filter")
+                    if filters not in (
+                        "/FlateDecode",
+                        "/Fl",
+                        ["/FlateDecode"],
+                        ["/Fl"],
+                    ):
+                        continue
+                    # pypdf's recovery decoder can silently return empty bytes
+                    # for an unreadable FlateDecode stream. Distinguish that
+                    # from a genuinely empty compressed stream, without treating
+                    # successful nonempty recovery as strict-conformance failure.
+                    if not part.get_data():
+                        encoded = getattr(part, "_data", None)
+                        if not isinstance(encoded, bytes):
+                            raise UncheckedArtifact(
+                                "PDF reader does not expose encoded stream bytes for recovery verification."
+                            )
+                        try:
+                            decoder = zlib.decompressobj()
+                            decoded = decoder.decompress(encoded, 1)
+                        except zlib.error as exc:
+                            raise UncheckedArtifact(
+                                "PDF content stream recovery could not be verified."
+                            ) from exc
+                        if decoded or not decoder.eof:
+                            raise UncheckedArtifact(
+                                "PDF content stream recovery could not be verified."
+                            )
+            stream = page.get_contents()
+            if stream is not None:
+                expanded += len(stream.get_data())
+                if expanded > content.limits.max_expanded_bytes:
+                    raise UncheckedArtifact("PDF exceeds the decoded content budget.")
+    except (PdfReadError, ValueError, KeyError, TypeError, OSError) as exc:
+        raise InvalidArtifact(
+            "PDF structure or page content cannot be decoded."
+        ) from exc
+    except (PyPdfError, DependencyError) as exc:
+        raise UncheckedArtifact(
+            "PDF reader could not complete validation (dependency or parser limit)."
+        ) from exc
+
+
+def check_image(content: ArtifactContent) -> None:
+    from PIL import Image, UnidentifiedImageError
+
+    try:
+        with Image.open(BytesIO(content.data)) as image:
+            width, height = image.size
+            frames = getattr(image, "n_frames", 1)
+            if width * height * frames > content.limits.max_pixels:
+                raise UncheckedArtifact("Image exceeds the decoded pixel budget.")
+            image.verify()
+        # verify() checks container structure; load() also exercises decoding.
+        with Image.open(BytesIO(content.data)) as image:
+            for index in range(getattr(image, "n_frames", 1)):
+                image.seek(index)
+                image.load()
+    except Image.DecompressionBombError as exc:
+        raise UncheckedArtifact("Image exceeds the decoded pixel budget.") from exc
+    except (UnidentifiedImageError, OSError, ValueError, SyntaxError) as exc:
+        raise InvalidArtifact("Image data cannot be decoded.") from exc
diff --git a/src/xagent/core/artifact_validation/models.py b/src/xagent/core/artifact_validation/models.py
new file mode 100644
index 0000000000..ef4f4a571d
--- /dev/null
+++ b/src/xagent/core/artifact_validation/models.py
@@ -0,0 +1,69 @@
+"""Transport-neutral contracts for bounded, format-specific artifact checks."""
+
+from dataclasses import dataclass
+from typing import Any, Callable, Literal
+
+ValidationStatus = Literal["valid", "invalid", "unchecked"]
+
+
+class InvalidArtifact(Exception):
+    """A known format error; the message must be safe for user/model output."""
+
+
+class UncheckedArtifact(Exception):
+    """The check cannot establish validity (dependency, budget, encryption)."""
+
+
+@dataclass(frozen=True)
+class ValidationLimits:
+    max_bytes: int = 32 * 1024 * 1024
+    max_expanded_bytes: int = 64 * 1024 * 1024
+    max_entries: int = 2048
+    max_units: int = 200_000
+    max_pixels: int = 25_000_000
+
+
+@dataclass(frozen=True)
+class ArtifactContent:
+    """An immutable snapshot. Checks must not read paths or change files."""
+
+    filename: str
+    data: bytes
+    limits: ValidationLimits
+
+
+@dataclass(frozen=True)
+class ArtifactCheck:
+    name: str
+    extensions: frozenset[str]
+    run: Callable[[ArtifactContent], None]
+
+
+@dataclass(frozen=True)
+class CheckResult:
+    name: str
+    status: ValidationStatus
+    message: str
+
+
+@dataclass(frozen=True)
+class ValidationReport:
+    status: ValidationStatus
+    checks: tuple[CheckResult, ...]
+    sha256: str | None = None
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "status": self.status,
+            "sha256": self.sha256,
+            "checks": [
+                {"name": c.name, "status": c.status, "message": c.message}
+                for c in self.checks
+            ],
+        }
+
+
+def unchecked(message: str, *, sha256: str | None = None) -> ValidationReport:
+    return ValidationReport(
+        "unchecked", (CheckResult("availability", "unchecked", message),), sha256
+    )
diff --git a/src/xagent/core/artifact_validation/office.py b/src/xagent/core/artifact_validation/office.py
new file mode 100644
index 0000000000..2f2b286645
--- /dev/null
+++ b/src/xagent/core/artifact_validation/office.py
@@ -0,0 +1,96 @@
+"""OOXML package preflight, followed by the actual document readers."""
+
+from io import BytesIO
+from pathlib import Path, PurePosixPath
+from zipfile import BadZipFile, ZipFile
+
+from .models import ArtifactContent, InvalidArtifact, UncheckedArtifact
+
+
+def check_package(content: ArtifactContent) -> None:
+    from xml.etree.ElementTree import ParseError
+
+    from defusedxml.common import DefusedXmlException
+    from defusedxml.ElementTree import fromstring
+
+    try:
+        with ZipFile(BytesIO(content.data)) as archive:
+            entries = archive.infolist()
+            if (
+                len(entries) > content.limits.max_entries
+                or sum(e.file_size for e in entries) > content.limits.max_expanded_bytes
+            ):
+                raise UncheckedArtifact("Office package exceeds the expansion budget.")
+            names = [e.filename for e in entries]
+            if len(names) != len(set(names)) or any(
+                PurePosixPath(n).is_absolute() or ".." in PurePosixPath(n).parts
+                for n in names
+            ):
+                raise InvalidArtifact("Office package contains ambiguous member paths.")
+            # Main-part locations come from content types and relationships,
+            # not conventional filenames. The format reader resolves them
+            # after this package-wide safety preflight has passed.
+            required = {"[Content_Types].xml", "_rels/.rels"}
+            if not required.issubset(names):
+                raise InvalidArtifact(
+                    "Office package is missing required document parts."
+                )
+            expanded = 0
+            for entry in entries:
+                if entry.flag_bits & 1:
+                    raise UncheckedArtifact(
+                        "Encrypted Office packages cannot be checked."
+                    )
+                # Read all members to exercise CRC/truncation checks, without
+                # extracting files or trusting compressed-size claims alone.
+                with archive.open(entry) as stream:
+                    data = stream.read(content.limits.max_expanded_bytes - expanded + 1)
+                expanded += len(data)
+                if expanded > content.limits.max_expanded_bytes:
+                    raise UncheckedArtifact(
+                        "Office package exceeds the expansion budget."
+                    )
+                if entry.filename.endswith((".xml", ".rels")):
+                    fromstring(data)
+    except (BadZipFile, ParseError, DefusedXmlException, EOFError, OSError) as exc:
+        raise InvalidArtifact(
+            "Office package is corrupt or contains unsafe XML."
+        ) from exc
+
+
+def check_office_document(content: ArtifactContent) -> None:
+    suffix = Path(content.filename).suffix.lower()
+    try:
+        if suffix == ".xlsx":
+            from openpyxl import load_workbook
+
+            workbook = load_workbook(
+                BytesIO(content.data), read_only=True, data_only=False, keep_links=False
+            )
+            try:
+                cells = 0
+                for sheet in workbook:
+                    # Ignore stale dimension metadata; walk actual cells. Empty
+                    # workbooks/templates are legal, not business failures.
+                    sheet.reset_dimensions()
+                    for row in sheet.iter_rows():
+                        cells += len(row)
+                        if cells > content.limits.max_units:
+                            raise UncheckedArtifact("Workbook exceeds the cell budget.")
+            finally:
+                workbook.close()
+        elif suffix == ".docx":
+            from docx import Document
+
+            document = Document(BytesIO(content.data))
+            _ = document.paragraphs, document.tables
+        else:
+            from pptx import Presentation
+
+            presentation = Presentation(BytesIO(content.data))
+            for slide in presentation.slides:
+                _ = slide.shapes
+    except (ValueError, KeyError, IndexError, TypeError, OSError, BadZipFile) as exc:
+        raise InvalidArtifact(
+            "Office document cannot be opened by its format reader."
+        ) from exc
diff --git a/src/xagent/core/artifact_validation/registry.py b/src/xagent/core/artifact_validation/registry.py
new file mode 100644
index 0000000000..7e31bfaa29
--- /dev/null
+++ b/src/xagent/core/artifact_validation/registry.py
@@ -0,0 +1,82 @@
+"""Checks compose by extension; delivery callers do not know about formats."""
+
+import hashlib
+import logging
+from pathlib import Path
+
+from .models import (
+    ArtifactCheck,
+    ArtifactContent,
+    CheckResult,
+    InvalidArtifact,
+    UncheckedArtifact,
+    ValidationReport,
+    unchecked,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ArtifactCheckRegistry:
+    def __init__(self) -> None:
+        self._checks: dict[str, ArtifactCheck] = {}
+
+    def register(self, check: ArtifactCheck) -> None:
+        if not check.name or check.name in self._checks:
+            raise ValueError(f"Duplicate or empty artifact check name: {check.name}")
+        if not check.extensions or any(
+            not ext.startswith(".") or ext != ext.lower() for ext in check.extensions
+        ):
+            raise ValueError("Check extensions must be nonempty lowercase suffixes")
+        self._checks[check.name] = check
+
+    def supports(self, filename: str) -> bool:
+        suffix = Path(filename).suffix.lower()
+        return any(suffix in check.extensions for check in self._checks.values())
+
+    @property
+    def extensions(self) -> frozenset[str]:
+        return frozenset(
+            ext for check in self._checks.values() for ext in check.extensions
+        )
+
+    def validate(self, content: ArtifactContent) -> ValidationReport:
+        digest = hashlib.sha256(content.data).hexdigest()
+        if len(content.data) > content.limits.max_bytes:
+            return unchecked("File exceeds the validation byte budget.", sha256=digest)
+        suffix = Path(content.filename).suffix.lower()
+        checks = [c for c in self._checks.values() if suffix in c.extensions]
+        if not checks:
+            return unchecked(
+                "No validator is installed for this format.", sha256=digest
+            )
+        results = []
+        for check in checks:
+            try:
+                check.run(content)
+            except InvalidArtifact as exc:
+                results.append(CheckResult(check.name, "invalid", str(exc)))
+            except UncheckedArtifact as exc:
+                results.append(CheckResult(check.name, "unchecked", str(exc)))
+            except ImportError:
+                results.append(
+                    CheckResult(
+                        check.name, "unchecked", "Parser dependency is unavailable."
+                    )
+                )
+            except Exception:
+                # A checker bug is not evidence of a corrupt user file. Never
+                # expose parser exceptions (which can contain paths/content).
+                logger.exception("Artifact check %s failed unexpectedly", check.name)
+                results.append(
+                    CheckResult(
+                        check.name, "unchecked", "Validator could not complete."
+                    )
+                )
+            else:
+                results.append(CheckResult(check.name, "valid", "Format check passed."))
+            if results[-1].status != "valid":
+                # Structural/budget preflights run before expensive decoders.
+                break
+        status = results[-1].status
+        return ValidationReport(status, tuple(results), digest)
diff --git a/src/xagent/core/artifact_validation/service.py b/src/xagent/core/artifact_validation/service.py
new file mode 100644
index 0000000000..7c9ea6d191
--- /dev/null
+++ b/src/xagent/core/artifact_validation/service.py
@@ -0,0 +1,164 @@
+"""Bounded host-side validation, cached by bytes rather than a mutable filename.
+
+Authorization/path ownership remains with the caller. This module neither
+registers nor deletes files and must never run inside a database transaction.
+"""
+
+import hashlib
+import json
+import logging
+import os
+import stat
+import subprocess
+import sys
+from collections import OrderedDict
+from pathlib import Path
+from threading import BoundedSemaphore, Lock
+
+from ...config import (
+    get_artifact_validation_max_bytes,
+    get_artifact_validation_timeout_seconds,
+    in_sandbox_tool_runner,
+)
+from .defaults import default_registry
+from .models import CheckResult, ValidationReport, unchecked
+
+logger = logging.getLogger(__name__)
+
+_slots = BoundedSemaphore(2)
+# Public capability URLs may be shared widely. Reserve at least one worker
+# slot for authenticated/tool callers and never queue public requests here.
+_public_slots = BoundedSemaphore(1)
+_cache_lock = Lock()
+_cache: OrderedDict[tuple[str, str, int], ValidationReport] = OrderedDict()
+
+
+def _run_checks(
+    filename: str, data: bytes, max_bytes: int, timeout: float
+) -> ValidationReport:
+    try:
+        # A subprocess timeout actually terminates a stuck parser; timing out a
+        # thread would leave it consuming resources. Bytes, not paths, cross
+        # this boundary, so the report describes exactly the parent snapshot.
+        completed = subprocess.run(
+            [
+                sys.executable,
+                "-m",
+                "xagent.core.artifact_validation.worker",
+                filename,
+                str(max_bytes),
+            ],
+            input=data,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.DEVNULL,
+            timeout=timeout,
+            check=True,
+            env={**os.environ, "OPENBLAS_NUM_THREADS": "1"},
+        )
+        value = json.loads(completed.stdout)
+        statuses = {"valid", "invalid", "unchecked"}
+        if value["status"] not in statuses or not value["checks"]:
+            raise ValueError("Invalid validation report")
+        checks = tuple(CheckResult(**item) for item in value["checks"])
+        if any(c.status not in statuses for c in checks):
+            raise ValueError("Invalid check status")
+        return ValidationReport(value["status"], checks, value["sha256"])
+    except subprocess.TimeoutExpired:
+        return unchecked("File validation exceeded its time budget.")
+    except (subprocess.SubprocessError, OSError, ValueError, KeyError, TypeError):
+        logger.exception("Artifact validator process failed unexpectedly")
+        return unchecked("Validator process could not complete.")
+
+
+def validate_artifact(
+    path: str | Path, *, filename: str | None = None, public: bool = False
+) -> ValidationReport:
+    if in_sandbox_tool_runner():
+        return unchecked("Awaiting host-side file validation.")
+    path = Path(path)
+    filename = Path(filename or path.name).name
+    if not default_registry().supports(filename):
+        return unchecked("No validator is installed for this format.")
+    try:
+        max_bytes = get_artifact_validation_max_bytes()
+        timeout = get_artifact_validation_timeout_seconds()
+    except ValueError:
+        logger.warning("Invalid artifact validation configuration; checks are disabled")
+        return unchecked("Validation configuration is invalid.")
+    if public and not _public_slots.acquire(blocking=False):
+        return unchecked("Public validation capacity is busy; retry later.")
+    # Acquire before loading bytes, not merely before launching the parser.
+    # Concurrent preview requests must not each retain a max-sized snapshot.
+    try:
+        if not _slots.acquire(timeout=timeout):
+            return unchecked("Validation capacity is busy; file has not been checked.")
+        try:
+            try:
+                return _validate_snapshot(path, filename, max_bytes, timeout)
+            except Exception:
+                # Validation is advisory: snapshot/host failures must not turn
+                # a successfully registered output into a failed tool result.
+                logger.exception("Artifact snapshot validation failed unexpectedly")
+                return unchecked("File validation could not complete.")
+        finally:
+            _slots.release()
+    finally:
+        if public:
+            _public_slots.release()
+
+
+def _validate_snapshot(
+    path: Path, filename: str, max_bytes: int, timeout: float
+) -> ValidationReport:
+    try:
+        if path.stat().st_size > max_bytes:
+            return unchecked("File exceeds the validation byte budget.")
+        # Nonblocking open prevents a replaced file/FIFO from hanging before
+        # the killable parser process even starts. Validate the opened inode.
+        with os.fdopen(
+            os.open(path, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0)), "rb"
+        ) as stream:
+            before = os.fstat(stream.fileno())
+            if not stat.S_ISREG(before.st_mode):
+                return unchecked("Only regular files can be validated.")
+            data = stream.read(max_bytes + 1)
+            after = os.fstat(stream.fileno())
+        current = path.stat()
+    except OSError:
+        return unchecked("File bytes are unavailable for validation.")
+    if len(data) > max_bytes:
+        return unchecked("File exceeds the validation byte budget.")
+
+    def identity(stat_result: os.stat_result) -> tuple[int, ...]:
+        return (
+            stat_result.st_dev,
+            stat_result.st_ino,
+            stat_result.st_size,
+            stat_result.st_mtime_ns,
+            stat_result.st_ctime_ns,
+        )
+
+    if identity(before) != identity(after) or identity(after) != identity(current):
+        return unchecked("File changed during validation; check its new version.")
+    digest = hashlib.sha256(data).hexdigest()
+    key = (digest, filename, max_bytes)
+    with _cache_lock:
+        cached = _cache.get(key)
+        if cached:
+            _cache.move_to_end(key)
+    report = cached or _run_checks(filename, data, max_bytes, timeout)
+    try:
+        if identity(path.stat()) != identity(after):
+            return unchecked("File changed during validation; check its new version.")
+    except OSError:
+        return unchecked("File bytes are unavailable for validation.")
+    if report.sha256 not in (None, digest):
+        return unchecked("Validation report did not match the file snapshot.")
+    report = ValidationReport(report.status, report.checks, digest)
+    if report.status != "unchecked":
+        with _cache_lock:
+            _cache[key] = report
+            _cache.move_to_end(key)
+            while len(_cache) > 128:
+                _cache.popitem(last=False)
+    return report
diff --git a/src/xagent/core/artifact_validation/worker.py b/src/xagent/core/artifact_validation/worker.py
new file mode 100644
index 0000000000..443cc36a12
--- /dev/null
+++ b/src/xagent/core/artifact_validation/worker.py
@@ -0,0 +1,37 @@
+"""Short-lived parser process. Input bytes are snapshotted by the parent."""
+
+import json
+import sys
+
+from .defaults import default_registry
+from .models import ArtifactContent, ValidationLimits
+
+
+def main() -> None:
+    # Bound decoder allocations on Linux in addition to parent-enforced wall
+    # time and per-format expansion limits. macOS does not enforce RLIMIT_AS
+    # consistently, so retain the portable input/expansion/pixel budgets there.
+    if sys.platform == "linux":
+        import resource
+
+        soft, hard = resource.getrlimit(resource.RLIMIT_AS)
+        resource.setrlimit(
+            resource.RLIMIT_AS,
+            (
+                1024**3 if soft == resource.RLIM_INFINITY else min(soft, 1024**3),
+                1024**3 if hard == resource.RLIM_INFINITY else min(hard, 1024**3),
+            ),
+        )
+    limits = ValidationLimits(max_bytes=int(sys.argv[2]))
+    # Parser libraries may write warnings to stdout. Keep the protocol output
+    # separate from their diagnostics.
+    output = sys.stdout
+    sys.stdout = sys.stderr
+    data = sys.stdin.buffer.read(limits.max_bytes + 1)
+    report = default_registry().validate(ArtifactContent(sys.argv[1], data, limits))
+    output.write(json.dumps(report.as_dict()))
+    output.flush()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/xagent/core/file_ref.py b/src/xagent/core/file_ref.py
index a00aa9d1ea..972ae1eb50 100644
--- a/src/xagent/core/file_ref.py
+++ b/src/xagent/core/file_ref.py
@@ -75,6 +75,10 @@ def final_deliverable_file_reference_instructions(
 - When the user requests a new file or file-based artifact, it is not delivered until a successful tool result returns its registered FileRef or markdown_link.
 - Do not call final_answer claiming that a file was created or delivered unless that result exists."""
 
+FILE_REF_OUTPUT_INSTRUCTIONS += """
+- Tool execution success and file validation are separate. If validation.status is invalid, repair the file and recheck it, or clearly report the failure; do not present it as a usable completed deliverable. Keep its file_id available for repair.
+- An unchecked or absent validation result is not a pass. State that limitation when delivering the file; a valid result establishes format readability only, not business/content correctness."""
+
 FILE_REF_MODEL_INSTRUCTIONS = f"""## FILE REFERENCES
 Files are referenced by FileRef objects. Treat file_id as the canonical file handle.
 
@@ -192,12 +196,17 @@ def build_workspace_file_ref(
     file_id: str | None = None,
     mime_type: str | None = None,
     internal: bool = False,
+    validate: bool = False,
 ) -> dict[str, Any]:
     """Register a workspace file and build the model/API-facing FileRef.
 
     ``internal`` keeps execution scratch data resolvable by ``file_id`` without
     creating a user-visible or durably uploaded file record. It fails closed
     when the workspace does not support internal registration.
+
+    ``validate`` checks a completed output snapshot without changing registration
+    or tool success. Call from a worker thread, as with other blocking file I/O.
+    It applies only to user-visible outputs; internal scratch files are not checked.
     """
     resolved_path = Path(file_path).resolve()
     if not resolved_path.exists() or not resolved_path.is_file():
@@ -233,6 +242,10 @@ def build_workspace_file_ref(
         except ValueError:
             relative_path = str(resolved_path)
         result["relative_path"] = relative_path
+        if validate:
+            from .artifact_validation.service import validate_artifact
+
+            result["validation"] = validate_artifact(resolved_path).as_dict()
     return result
 
 
diff --git a/src/xagent/core/tools/adapters/vibe/sandboxed_tool/sandboxed_tool_wrapper.py b/src/xagent/core/tools/adapters/vibe/sandboxed_tool/sandboxed_tool_wrapper.py
index 654f3567c8..502b2f8770 100644
--- a/src/xagent/core/tools/adapters/vibe/sandboxed_tool/sandboxed_tool_wrapper.py
+++ b/src/xagent/core/tools/adapters/vibe/sandboxed_tool/sandboxed_tool_wrapper.py
@@ -493,6 +493,16 @@ def run_json_sync(self, args: Mapping[str, Any]) -> Any:
     async def run_json_async(self, args: Mapping[str, Any]) -> Any:
         """Execute the tool in the sandbox, then register its files here."""
         result = await self._run_json_in_sandbox(args)
+        # Guest metadata is not evidence about persisted host bytes. Only the
+        # host rebuild below can attach an authoritative validation report.
+        pending = [result]
+        while pending:
+            value = pending.pop()
+            if isinstance(value, dict):
+                value.pop("validation", None)
+                pending.extend(value.values())
+            elif isinstance(value, list):
+                pending.extend(value)
         try:
             return await self._register_sandbox_outputs(result)
         except Exception:
diff --git a/src/xagent/core/tools/artifacts.py b/src/xagent/core/tools/artifacts.py
index a95153f150..04bef4458f 100644
--- a/src/xagent/core/tools/artifacts.py
+++ b/src/xagent/core/tools/artifacts.py
@@ -6,6 +6,7 @@
 from pathlib import Path
 from typing import Any, Iterable
 
+from ..artifact_validation.defaults import default_registry
 from ..file_ref import (
     build_workspace_file_ref,
     guess_mime_type,
@@ -33,7 +34,7 @@
     ".webm",
     ".xls",
     ".xlsx",
-}
+} | set(default_registry().extensions)
 SAFE_FILE_REF_KEYS = {
     "download_url",
     "file_id",
@@ -44,6 +45,7 @@
     "preview_url",
     "relative_path",
     "size",
+    "validation",
 }
 # Every media tool returns its artifact as <kind>_path alongside file_id/file_ref.
 # Missing keys were redacted only by coincidence: file_ref.file_path registers the
@@ -100,7 +102,7 @@ def artifact_type_for_filename(filename: str) -> str:
     return "file"
 
 
-def build_inline_artifact(file_ref: dict[str, Any]) -> dict[str, str]:
+def build_inline_artifact(file_ref: dict[str, Any]) -> dict[str, Any]:
     filename = str(file_ref.get("filename") or "artifact")
     return {
         "type": artifact_type_for_filename(filename),
@@ -108,6 +110,7 @@ def build_inline_artifact(file_ref: dict[str, Any]) -> dict[str, str]:
         "filename": filename,
         "mime_type": str(file_ref.get("mime_type") or guess_mime_type(filename)),
         "display": "inline",
+        **({"validation": file_ref["validation"]} if "validation" in file_ref else {}),
     }
 
 
@@ -148,7 +151,7 @@ def format_tool_result_for_observation(tool_name: str, result: Any) -> str:
 
     metadata = _observation_metadata(sanitized, _OBSERVATION_EXCLUDED_KEYS)
     return (
-        f"Tool '{tool_name}' produced displayable artifact(s):\n"
+        f"Tool '{tool_name}' produced artifact(s); check validation before delivery:\n"
         + "\n".join(artifact_lines)
         + "\nUse the Markdown/chat form in assistant messages. "
         + "When writing HTML for Xagent preview, reference the same file_id "
@@ -179,7 +182,7 @@ def _format_artifact_lines(artifacts: list[Any]) -> list[str]:
                     [
                         f"- {filename}",
                         "  file_id: unavailable, registration did not complete",
-                        "  The file itself is written and intact. No file_id "
+                        "  The tool reported a written file; this does not establish readability. No file_id "
                         "can be obtained for it in this task; say so plainly "
                         "and never rewrite the file to try to mint one.",
                     ]
@@ -200,6 +203,33 @@ def _format_artifact_lines(artifacts: list[Any]) -> list[str]:
                 ]
             )
         )
+        validation = artifact.get("validation")
+        if not isinstance(validation, dict):
+            lines.append(
+                "  Validation: NOT RUN. No validation report was produced for this file."
+            )
+            continue
+        status = validation.get("status", "unchecked")
+        if status == "invalid":
+            lines.append(
+                "  Validation: INVALID. Repair and recheck, or report failure; do not claim a usable deliverable."
+            )
+        elif status == "valid":
+            lines.append(
+                "  Validation: format readable; content correctness is not checked."
+            )
+        else:
+            lines.append(
+                "  Validation: UNCHECKED. Do not claim this file passed validation."
+            )
+            checks = validation.get("checks")
+            if isinstance(checks, list):
+                for check in checks:
+                    if isinstance(check, dict) and check.get("status") == "unchecked":
+                        message = check.get("message")
+                        if isinstance(message, str) and message.strip():
+                            lines.append(f"  Validation reason: {message}")
+                            break
     return lines
 
 
@@ -304,7 +334,7 @@ def build_generated_file_metadata(
     file_paths: Iterable[str | Path],
 ) -> dict[str, list[Any]]:
     file_refs: list[dict[str, Any]] = []
-    artifacts: list[dict[str, str]] = []
+    artifacts: list[dict[str, Any]] = []
     generated_files: list[str] = []
 
     for file_path in sorted({Path(path).resolve() for path in file_paths}):
@@ -314,7 +344,7 @@ def build_generated_file_metadata(
             continue
         try:
             file_ref = build_workspace_file_ref(
-                workspace=workspace, file_path=file_path
+                workspace=workspace, file_path=file_path, validate=True
             )
         except Exception as exc:  # noqa: BLE001
             logger.warning(
diff --git a/src/xagent/core/tools/core/workspace_file_tool.py b/src/xagent/core/tools/core/workspace_file_tool.py
index 51278ad6ff..e3d7d766a3 100644
--- a/src/xagent/core/tools/core/workspace_file_tool.py
+++ b/src/xagent/core/tools/core/workspace_file_tool.py
@@ -364,6 +364,7 @@ def write_file(
         file_ref = build_workspace_file_ref(
             workspace=self.workspace,
             file_path=resolved_path,
+            validate=True,
         )
 
         logger.debug(
@@ -381,6 +382,7 @@ def _registered_write_result(self, file_path: Path) -> Dict[str, Any]:
         file_ref = build_workspace_file_ref(
             workspace=self.workspace,
             file_path=file_path,
+            validate=True,
         )
         return {
             "success": True,
diff --git a/src/xagent/web/api/files.py b/src/xagent/web/api/files.py
index ed1cc0def1..ba71cd1f6e 100644
--- a/src/xagent/web/api/files.py
+++ b/src/xagent/web/api/files.py
@@ -19,6 +19,7 @@
 )
 from fastapi.responses import (
     FileResponse,
+    JSONResponse,
     RedirectResponse,
     Response,
 )
@@ -36,6 +37,8 @@
     get_storage_root,
     get_uploads_dir,
 )
+from ...core.artifact_validation.defaults import default_registry
+from ...core.artifact_validation.service import validate_artifact
 from ...core.execution_scope import resolve_execution_scope
 from ...core.file_storage import get_user_file_storage, normalize_storage_key
 from ...core.tools.adapters.vibe.file_tool import read_file
@@ -233,7 +236,9 @@ async def _inline_preview_response(
     media_type: str,
     file_id: Optional[str] = None,
     extra_headers: Optional[Dict[str, str]] = None,
-) -> FileResponse:
+    validation_only: bool = False,
+    public_validation: bool = False,
+) -> Response:
     """Build the final inline preview FileResponse, rasterizing SVG to PNG.
 
     Raw SVG bytes are never served inline — an embedded ``<script>`` would
@@ -248,6 +253,16 @@ async def _inline_preview_response(
     on normal browser caching for repeatedly-rendered chat images.
     """
 
+    if validation_only:
+        report = await asyncio.to_thread(
+            validate_artifact, path, filename=filename, public=public_validation
+        )
+        return JSONResponse(
+            {**report.as_dict(), "supported": default_registry().supports(filename)},
+            media_type="application/vnd.xagent.validation+json",
+            headers={"Cache-Control": "no-store"},
+        )
+
     if media_type == "image/svg+xml":
         try:
             # CairoSVG rasterization is CPU-bound and can run up to the
@@ -1763,6 +1778,7 @@ async def preview_file(
     file_id: str,
     auth: _AuthenticatedFileRequest = Depends(_user_from_bearer_or_stream_ticket),
     db: Session = Depends(get_db),
+    validation_only: bool = False,
 ) -> Any:
     user, via_stream_ticket = auth
     cache_headers = _PUBLIC_PREVIEW_CACHE_HEADERS if via_stream_ticket else None
@@ -1797,7 +1813,7 @@ async def preview_file(
         # hop, signed-URL TTL thereafter), and an operator tuning either
         # XAGENT_*_TTL_SECONDS knob in isolation can unexpectedly cut
         # playback off at the other one.
-        if _preview_can_redirect(full_path, media_type):
+        if not validation_only and _preview_can_redirect(full_path, media_type):
             redirect_response = _durable_redirect_response(
                 file_ref,
                 filename=file_name,
@@ -1830,12 +1846,16 @@ async def preview_file(
             except DurableObjectMissingError:
                 materialized_path = file_ref.local_path
                 _ensure_under_uploads(materialized_path, owner_user_id)
+            if validation_only:
+                release_db_connection_if_clean(db)
             return await _inline_preview_response(
                 materialized_path,
                 filename=file_name,
                 media_type=media_type,
                 file_id=file_id if is_valid_uuid(file_id) else None,
                 extra_headers=cache_headers,
+                validation_only=validation_only,
+                public_validation=via_stream_ticket,
             )
     else:
         # For legacy files without records, check ownership
@@ -1849,7 +1869,7 @@ async def preview_file(
     if not full_path.exists():
         raise HTTPException(status_code=404, detail="File not found")
 
-    if _preview_can_redirect(full_path, media_type):
+    if not validation_only and _preview_can_redirect(full_path, media_type):
         # See the comment on the equivalent branch above: not gated on
         # ticket presence, for the same reason.
         accel_response = _accel_redirect_response(
@@ -1863,12 +1883,16 @@ async def preview_file(
         if accel_response is not None:
             return accel_response
 
+    if validation_only:
+        release_db_connection_if_clean(db)
     return await _inline_preview_response(
         full_path,
         filename=file_name,
         media_type=media_type,
         file_id=file_id if is_valid_uuid(file_id) else None,
         extra_headers=cache_headers,
+        validation_only=validation_only,
+        public_validation=via_stream_ticket,
     )
 
 
@@ -2050,6 +2074,7 @@ async def public_preview_file(
     relative_path: Optional[str] = Query(default=None),
     token: Optional[str] = Query(default=None),
     db: Session = Depends(get_db),
+    validation_only: bool = False,
 ) -> Any:
     # For public preview, we need to handle both file_id and legacy paths
     # Try UUID first
@@ -2081,12 +2106,17 @@ async def public_preview_file(
             except DurableObjectMissingError:
                 target_path = file_ref.local_path
                 _ensure_under_uploads(target_path, owner_user_id)
+            filename = str(file_record.filename)
+            if validation_only:
+                release_db_connection_if_clean(db)
             return await _inline_preview_response(
                 target_path,
-                filename=str(file_record.filename),
-                media_type=guess_media_type(str(file_record.filename)),
+                filename=filename,
+                media_type=guess_media_type(filename),
                 file_id=file_id if is_valid_uuid(file_id) else None,
                 extra_headers=_PUBLIC_PREVIEW_CACHE_HEADERS,
+                validation_only=validation_only,
+                public_validation=True,
             )
     else:
         # Try to resolve as legacy path across all user directories
@@ -2138,12 +2168,16 @@ async def public_preview_file(
     if not target_path.exists() or not target_path.is_file():
         raise HTTPException(status_code=404, detail="File not found")
 
+    if validation_only:
+        release_db_connection_if_clean(db)
     return await _inline_preview_response(
         target_path,
         filename=target_path.name,
         media_type=guess_media_type(target_path.name),
         file_id=file_id if is_valid_uuid(file_id) else None,
         extra_headers=_PUBLIC_PREVIEW_CACHE_HEADERS,
+        validation_only=validation_only,
+        public_validation=True,
     )
 
 
diff --git a/tests/core/test_artifact_validation.py b/tests/core/test_artifact_validation.py
new file mode 100644
index 0000000000..d3951d5095
--- /dev/null
+++ b/tests/core/test_artifact_validation.py
@@ -0,0 +1,701 @@
+"""Format checks are composable and never conflate execution with delivery."""
+
+import hashlib
+import json
+import logging
+import os
+import re
+import struct
+import subprocess
+import sys
+from io import BytesIO
+from unittest.mock import Mock
+from zipfile import ZIP_DEFLATED, ZipFile
+
+import pytest
+
+from xagent.core.artifact_validation import (
+    ArtifactCheck,
+    ArtifactCheckRegistry,
+    ArtifactContent,
+    ValidationLimits,
+    service,
+)
+from xagent.core.artifact_validation.defaults import default_registry
+from xagent.core.artifact_validation.models import InvalidArtifact, UncheckedArtifact
+from xagent.core.artifact_validation.service import validate_artifact
+
+
+def check(data, filename, **limits):
+    return default_registry().validate(
+        ArtifactContent(filename, data, ValidationLimits(**limits))
+    )
+
+
+@pytest.mark.parametrize("extension", ["xlsx", "docx", "pptx", "pdf", "png"])
+def test_corrupt_formats_are_invalid(extension):
+    assert check(b"not a document", f"report.{extension}").status == "invalid"
+
+
+@pytest.mark.parametrize(
+    "extension", ["xlsx", "docx", "pptx", "pdf", "png", "bmp", "tif", "tiff"]
+)
+def test_real_readers_accept_minimal_documents(extension):
+    stream = BytesIO()
+    if extension == "xlsx":
+        from openpyxl import Workbook
+
+        Workbook().save(stream)
+    elif extension == "docx":
+        from docx import Document
+
+        Document().save(stream)
+    elif extension == "pptx":
+        from pptx import Presentation
+
+        Presentation().save(stream)
+    elif extension == "pdf":
+        from pypdf import PdfWriter
+
+        writer = PdfWriter()
+        writer.add_blank_page(width=100, height=100)
+        writer.write(stream)
+    else:
+        from PIL import Image
+
+        Image.new("RGB", (4, 4)).save(
+            stream,
+            format={"tif": "TIFF", "tiff": "TIFF"}.get(extension, extension.upper()),
+        )
+    result = check(stream.getvalue(), f"file.{extension}")
+    assert result.status == "valid", result
+
+
+@pytest.mark.parametrize(
+    "extension,part",
+    [
+        ("xlsx", "xl/workbook.xml"),
+        ("docx", "word/document.xml"),
+        ("pptx", "ppt/presentation.xml"),
+    ],
+)
+@pytest.mark.parametrize("missing", [False, True])
+def test_office_reader_resolves_main_part(extension, part, missing):
+    from xagent.core.artifact_validation.office import check_office_document
+
+    original = BytesIO()
+    if extension == "xlsx":
+        from openpyxl import Workbook
+
+        Workbook().save(original)
+    elif extension == "docx":
+        from docx import Document
+
+        Document().save(original)
+    else:
+        from pptx import Presentation
+
+        Presentation().save(original)
+    renamed = part.replace(".xml", "-renamed.xml")
+    old_rels = part.rsplit("/", 1)[0] + "/_rels/" + part.rsplit("/", 1)[1] + ".rels"
+    new_rels = old_rels.replace(".xml.rels", "-renamed.xml.rels")
+    rewritten = BytesIO()
+    with ZipFile(original) as source, ZipFile(rewritten, "w", ZIP_DEFLATED) as target:
+        for entry in source.infolist():
+            if missing and entry.filename == part:
+                continue
+            data = source.read(entry)
+            name = entry.filename
+            if not missing:
+                if name in ("[Content_Types].xml", "_rels/.rels"):
+                    data = data.replace(part.encode(), renamed.encode())
+                name = {part: renamed, old_rels: new_rels}.get(name, name)
+            target.writestr(name, data)
+    data = rewritten.getvalue()
+    if not missing:
+        # Prove that the installed format reader accepts the renamed package.
+        check_office_document(
+            ArtifactContent("file." + extension, data, ValidationLimits())
+        )
+    report = check(data, "file." + extension)
+    assert [c.status for c in report.checks] == [
+        "valid",
+        "invalid" if missing else "valid",
+    ]
+
+
+@pytest.mark.parametrize("extension,format_name", [("gif", "GIF"), ("tiff", "TIFF")])
+def test_image_checks_all_frames(extension, format_name, monkeypatch):
+    from PIL import Image
+
+    stream = BytesIO()
+    Image.new("RGB", (4, 4), "red").save(
+        stream,
+        format=format_name,
+        save_all=True,
+        append_images=[Image.new("RGB", (4, 4), "blue")],
+    )
+    data = stream.getvalue()
+    assert check(data, "frames." + extension).status == "valid"
+    assert check(data, "frames." + extension, max_pixels=31).status == "unchecked"
+    with Image.open(BytesIO(data)) as image:
+        reader_class = type(image)
+    original_load = reader_class.load
+
+    def fail_second_frame(self, *args, **kwargs):
+        if self.tell() == 1:
+            raise OSError("second frame cannot decode")
+        return original_load(self, *args, **kwargs)
+
+    monkeypatch.setattr(reader_class, "load", fail_second_frame)
+    assert check(data, "frames." + extension).status == "invalid"
+
+
+def test_read_only_xlsx_checks_cells_beyond_stale_dimensions():
+    from openpyxl import Workbook
+
+    original = BytesIO()
+    workbook = Workbook()
+    sheet = workbook.active
+    sheet.append(["header"])
+    sheet.append([1])
+    sheet.append([123])
+    workbook.save(original)
+
+    rewritten = BytesIO()
+    with ZipFile(original) as source, ZipFile(rewritten, "w", ZIP_DEFLATED) as target:
+        for entry in source.infolist():
+            data = source.read(entry)
+            if entry.filename == "xl/worksheets/sheet1.xml":
+                assert b'<dimension ref="A1:A3"/>' in data
+                assert b"<v>123</v>" in data
+                data = data.replace(
+                    b'<dimension ref="A1:A3"/>', b'<dimension ref="A1:A1"/>'
+                )
+                data = data.replace(b"<v>123</v>", b"<v>not-a-number</v>")
+            target.writestr(entry, data)
+
+    # The XML/ZIP remains well formed. Without resetting the read-only sheet's
+    # bounds, iteration stops at row 2 before decoding the corrupt third row.
+    report = check(rewritten.getvalue(), "stale-dimensions.xlsx")
+    assert [c.status for c in report.checks] == ["valid", "invalid"]
+
+
+@pytest.mark.parametrize(
+    "data",
+    [
+        b"",
+        b"a,b\n1,2\n",
+        b"one column",
+        b"a;b\n1;2",
+        b"a,b\n1,2,3",
+        "名字,值\n甲,1".encode("utf-16"),
+    ],
+)
+def test_csv_does_not_impose_business_schema(data):
+    assert check(data, "data.CSV").status == "valid"
+
+
+def test_csv_malformed_encoding_and_budget():
+    assert check(b'a,b\n"unterminated,2', "data.csv").status == "unchecked"
+    assert check(b"\xff\x98", "data.csv").status == "unchecked"
+    assert check(b"a,\x00", "data.csv").status == "invalid"
+    assert check(b"a\nb", "data.csv", max_units=1).status == "unchecked"
+    assert check(b"one", "data.csv", max_bytes=2).status == "unchecked"
+    assert check(b"whatever", "data.unknown").status == "unchecked"
+
+
+def test_csv_lenient_reader_recovery_is_not_reported_as_corruption():
+    import csv
+
+    data = b'a,b\n1,"he said "hi" there"\n'
+    assert len(list(csv.reader(data.decode().splitlines()))) == 2
+    assert check(data, "data.csv").status == "unchecked"
+
+
+@pytest.mark.parametrize(
+    "existing",
+    [(-1, -1), (512 * 1024**2, 512 * 1024**2), (256 * 1024**2, 512 * 1024**2)],
+)
+def test_worker_keeps_stricter_address_space_limits(existing, monkeypatch):
+    from io import StringIO
+    from types import SimpleNamespace
+
+    from xagent.core.artifact_validation import worker
+
+    resource = pytest.importorskip("resource")
+    monkeypatch.setattr(resource, "RLIM_INFINITY", -1)
+    applied = []
+
+    def set_limit(kind, limits):
+        assert kind == resource.RLIMIT_AS
+        if existing[1] != -1 and limits[1] > existing[1]:
+            raise ValueError("cannot raise the hard limit")
+        applied.append(limits)
+
+    monkeypatch.setattr(resource, "getrlimit", lambda _: existing)
+    monkeypatch.setattr(resource, "setrlimit", set_limit)
+    monkeypatch.setattr(sys, "platform", "linux")
+    monkeypatch.setattr(sys, "argv", ["worker", "data.csv", "1024"])
+    monkeypatch.setattr(sys, "stdin", SimpleNamespace(buffer=BytesIO(b"a,b\n1,2\n")))
+    output = StringIO()
+    flush = Mock(wraps=output.flush)
+    monkeypatch.setattr(output, "flush", flush)
+    monkeypatch.setattr(sys, "stdout", output)
+    worker.main()
+    flush.assert_called_once_with()
+    assert applied == [
+        tuple(1024**3 if limit == -1 else min(limit, 1024**3) for limit in existing)
+    ]
+    assert json.loads(output.getvalue())["status"] == "valid"
+
+
+def test_archive_budgets_and_unsafe_xml():
+    stream = BytesIO()
+    with ZipFile(stream, "w", ZIP_DEFLATED) as archive:
+        for name in ["[Content_Types].xml", "_rels/.rels", "xl/workbook.xml"]:
+            archive.writestr(name, "<a/>" * 20)
+    assert (
+        check(stream.getvalue(), "data.xlsx", max_expanded_bytes=10).status
+        == "unchecked"
+    )
+    stream = BytesIO()
+    with ZipFile(stream, "w") as archive:
+        for name in ["[Content_Types].xml", "_rels/.rels", "xl/workbook.xml"]:
+            archive.writestr(name, '<!DOCTYPE a [<!ENTITY x "value">]><a>&x;</a>')
+    assert check(stream.getvalue(), "data.xlsx").status == "invalid"
+
+
+@pytest.mark.parametrize(
+    "suffix,root",
+    [
+        ("xlsx", "xl/workbook.xml"),
+        ("docx", "word/document.xml"),
+        ("pptx", "ppt/presentation.xml"),
+    ],
+)
+@pytest.mark.parametrize(
+    "bad_member",
+    ["/absolute.xml", "../outside.xml", "xl/../../outside.xml", "[Content_Types].xml"],
+)
+def test_office_package_rejects_ambiguous_members(suffix, root, bad_member):
+    stream = BytesIO()
+    with ZipFile(stream, "w") as archive:
+        for name in ["[Content_Types].xml", "_rels/.rels", root, bad_member]:
+            archive.writestr(name, "<a/>")
+    report = check(stream.getvalue(), f"file.{suffix}")
+    assert report.status == "invalid"
+    assert len(report.checks) == 1
+    assert "ambiguous member paths" in report.checks[0].message
+
+
+@pytest.mark.parametrize("missing", ["[Content_Types].xml", "_rels/.rels"])
+def test_office_package_rejects_missing_required_parts(missing):
+    stream = BytesIO()
+    with ZipFile(stream, "w") as archive:
+        for name in ["[Content_Types].xml", "_rels/.rels", "xl/workbook.xml"]:
+            if name != missing:
+                archive.writestr(name, "<a/>")
+    report = check(stream.getvalue(), "file.xlsx")
+    assert report.status == "invalid"
+    assert "missing required document parts" in report.checks[0].message
+
+
+def test_office_package_entry_count_and_encrypted_member_guards():
+    stream = BytesIO()
+    with ZipFile(stream, "w") as archive:
+        for name in ["[Content_Types].xml", "_rels/.rels", "xl/workbook.xml"]:
+            archive.writestr(name, "<a/>")
+    report = check(stream.getvalue(), "file.xlsx", max_entries=2)
+    assert report.status == "unchecked"
+    assert "expansion budget" in report.checks[0].message
+
+    data = bytearray(stream.getvalue())
+    # Mark the first local and central-directory entry encrypted. The guard
+    # must stop before attempting to decode its untrusted payload.
+    struct.pack_into("<H", data, data.index(b"PK\x03\x04") + 6, 1)
+    struct.pack_into("<H", data, data.index(b"PK\x01\x02") + 8, 1)
+    report = check(bytes(data), "file.xlsx")
+    assert report.status == "unchecked"
+    assert "Encrypted Office" in report.checks[0].message
+
+
+@pytest.mark.parametrize(
+    "payload,expected", [(b"", "valid"), (b"q Q\n", "valid"), (None, "unchecked")]
+)
+@pytest.mark.parametrize("array", [False, True])
+def test_pdf_does_not_certify_silently_discarded_flate_content(
+    payload, expected, array
+):
+    import zlib
+
+    from pypdf import PdfWriter
+    from pypdf.generic import ArrayObject, EncodedStreamObject, NameObject
+
+    writer = PdfWriter()
+    page = writer.add_blank_page(width=100, height=100)
+    stream = EncodedStreamObject()
+    stream[NameObject("/Filter")] = NameObject("/FlateDecode")
+    stream._data = (
+        zlib.compress(payload) if payload is not None else b"not compressed data"
+    )
+    # A checksum defect can still recover every original content byte. Do not
+    # reject successful nonempty recovery just for strict zlib conformance.
+    if payload:
+        stream._data = stream._data[:-1] + bytes([stream._data[-1] ^ 1])
+        assert stream.get_data() == payload
+    ref = writer._add_object(stream)
+    page[NameObject("/Contents")] = ArrayObject([ref]) if array else ref
+    output = BytesIO()
+    writer.write(output)
+    report = check(output.getvalue(), "content.pdf")
+    assert report.status == expected
+    if expected == "unchecked":
+        assert "stream recovery" in report.checks[0].message
+
+
+@pytest.mark.parametrize(
+    "pages,max_units,expected",
+    [(500, 200_000, "valid"), (501, 200_000, "unchecked"), (3, 2, "unchecked")],
+)
+def test_pdf_page_ceiling_is_not_partial_success(pages, max_units, expected):
+    from pypdf import PdfWriter
+
+    writer = PdfWriter()
+    for _ in range(pages):
+        writer.add_blank_page(width=10, height=10)
+    output = BytesIO()
+    writer.write(output)
+    report = check(output.getvalue(), "pages.pdf", max_units=max_units)
+    assert report.status == expected
+    if expected == "unchecked":
+        assert f"{min(max_units, 500)} pages" in report.checks[-1].message
+
+
+@pytest.mark.parametrize("error_name", ["LimitReachedError", "DependencyError"])
+def test_pdf_reader_limits_and_dependencies_are_expected_unchecked(
+    monkeypatch, caplog, error_name
+):
+    import pypdf
+    from pypdf import errors
+
+    error_type = getattr(errors, error_name)
+    monkeypatch.setattr(
+        pypdf, "PdfReader", Mock(side_effect=error_type("private parser detail"))
+    )
+    report = check(b"%PDF-1.7\n", "limited.pdf")
+    assert report.status == "unchecked"
+    assert "dependency or parser limit" in report.checks[-1].message
+    assert "private parser detail" not in str(report.as_dict())
+    assert not caplog.records
+
+
+def test_pdf_missing_encoded_stream_api_has_explicit_diagnostic(monkeypatch):
+    from types import SimpleNamespace
+
+    import pypdf
+    from pypdf.generic import EncodedStreamObject, NameObject
+
+    stream = EncodedStreamObject()
+    stream[NameObject("/Filter")] = NameObject("/FlateDecode")
+    stream._data = b""
+    del stream._data
+    monkeypatch.setattr(stream, "get_data", lambda: b"")
+    monkeypatch.setattr(
+        pypdf,
+        "PdfReader",
+        lambda *args, **kwargs: SimpleNamespace(
+            is_encrypted=False, pages=[{"/Contents": stream}]
+        ),
+    )
+    report = check(b"%PDF-1.7\n", "stream.pdf")
+    assert report.status == "unchecked"
+    assert "does not expose encoded stream bytes" in report.checks[-1].message
+
+
+def test_readable_pdf_with_recoverable_xref_is_not_invalid():
+    from pypdf import PdfReader, PdfWriter
+    from pypdf.errors import PdfReadError
+
+    stream = BytesIO()
+    writer = PdfWriter()
+    writer.add_blank_page(width=100, height=100)
+    writer.write(stream)
+    original = stream.getvalue()
+    match = re.search(rb"startxref\s+(\d+)", original)
+    assert match
+    data = (
+        original[: match.start(1)]
+        + str(int(match[1]) + 1).encode()
+        + original[match.end(1) :]
+    )
+    with pytest.raises(PdfReadError):
+        PdfReader(BytesIO(data), strict=True)
+    assert len(PdfReader(BytesIO(data), strict=False).pages) == 1
+    assert check(data, "recoverable.pdf").status == "valid"
+    assert check(b"%PDF-1.7\nnot a document", "broken.pdf").status == "invalid"
+
+
+def test_cache_keeps_filename_and_extension_with_identical_bytes(tmp_path, monkeypatch):
+    service._cache.clear()
+    path = tmp_path / "artifact.bin"
+    path.write_bytes(b"a,b\n1,2")
+    run = Mock(wraps=service._run_checks)
+    monkeypatch.setattr(service, "_run_checks", run)
+    for filename, status in [
+        ("one.csv", "valid"),
+        ("two.csv", "valid"),
+        ("one.pdf", "invalid"),
+    ]:
+        assert validate_artifact(path, filename=filename).status == status
+        assert validate_artifact(path, filename=filename).status == status
+    assert run.call_count == 3
+
+
+def test_image_pixel_budget_and_pdf_encryption():
+    from PIL import Image
+    from pypdf import PdfWriter
+
+    stream = BytesIO()
+    Image.new("RGB", (4, 4)).save(stream, format="PNG")
+    assert check(stream.getvalue(), "image.png", max_pixels=4).status == "unchecked"
+    stream = BytesIO()
+    writer = PdfWriter()
+    writer.add_blank_page(width=10, height=10)
+    writer.encrypt("password")
+    writer.write(stream)
+    assert check(stream.getvalue(), "private.pdf").status == "unchecked"
+
+
+@pytest.mark.parametrize(
+    ("error", "status"),
+    [
+        (InvalidArtifact("bad structure"), "invalid"),
+        (UncheckedArtifact("budget"), "unchecked"),
+        (ImportError("missing"), "unchecked"),
+        (RuntimeError("/secret/path"), "unchecked"),
+    ],
+)
+def test_additional_checks_and_fail_closed_preflight(error, status):
+    registry = ArtifactCheckRegistry()
+    first = Mock()
+    second = Mock(side_effect=error)
+    third = Mock()
+    for name, callback in [("first", first), ("second", second), ("third", third)]:
+        registry.register(ArtifactCheck(name, frozenset({".foo"}), callback))
+    result = registry.validate(ArtifactContent("new.foo", b"bytes", ValidationLimits()))
+    assert result.status == status
+    assert [c.name for c in result.checks] == ["first", "second"]
+    assert "/secret/path" not in str(result.as_dict())
+    third.assert_not_called()
+    with pytest.raises(ValueError):
+        registry.register(ArtifactCheck("first", frozenset({".foo"}), first))
+
+
+def test_service_real_worker_and_same_name_rewrite(tmp_path):
+    path = tmp_path / "test.csv"
+    path.write_bytes(b"a,b\n1,2")
+    first = validate_artifact(path)
+    assert first.status == "valid"
+    assert first.sha256 == hashlib.sha256(path.read_bytes()).hexdigest()
+    path.write_bytes(b'a,b\n"broken')
+    second = validate_artifact(path)
+    assert second.status == "unchecked"
+    assert second.sha256 != first.sha256
+    assert path.read_bytes() == b'a,b\n"broken'
+
+
+@pytest.mark.parametrize(
+    "error", [MemoryError("read failed"), RuntimeError("unexpected read failure")]
+)
+def test_snapshot_exception_does_not_drop_generated_attachment(
+    tmp_path, monkeypatch, caplog, error
+):
+    from contextlib import contextmanager
+    from types import SimpleNamespace
+
+    from xagent.core.tools.artifacts import build_generated_file_metadata
+
+    path = tmp_path / "output.csv"
+    path.write_bytes(b"a,b\n1,2\n")
+    original = service.os.fdopen
+
+    @contextmanager
+    def failing_open(*args, **kwargs):
+        with original(*args, **kwargs) as stream:
+            wrapped = Mock(wraps=stream)
+            wrapped.read.side_effect = error
+            yield wrapped
+
+    monkeypatch.setattr(service.os, "fdopen", failing_open)
+    workspace = SimpleNamespace(
+        workspace_dir=tmp_path, get_file_id_from_path=lambda _: "registered-file"
+    )
+    metadata = build_generated_file_metadata(workspace=workspace, file_paths=[path])
+    assert len(metadata["file_refs"]) == 1
+    assert metadata["file_refs"][0]["file_id"] == "registered-file"
+    assert metadata["file_refs"][0]["validation"]["status"] == "unchecked"
+    assert path.read_bytes() == b"a,b\n1,2\n"
+    assert "snapshot validation failed" in caplog.text
+    assert service._slots.acquire(blocking=False)
+    assert service._slots.acquire(blocking=False)
+    service._slots.release()
+    service._slots.release()
+
+
+def test_cache_is_byte_keyed_and_detects_mid_check_rewrite(tmp_path, monkeypatch):
+    service._cache.clear()
+    path = tmp_path / "test.csv"
+    path.write_bytes(b"a,b")
+    run = Mock(wraps=service._run_checks)
+    monkeypatch.setattr(service, "_run_checks", run)
+    assert validate_artifact(path).status == "valid"
+    assert validate_artifact(path).status == "valid"
+    assert run.call_count == 1
+    service._cache.clear()
+
+    def rewrite(filename, data, *_args):
+        path.write_bytes(b"changed")
+        return check(data, filename)
+
+    monkeypatch.setattr(service, "_run_checks", rewrite)
+    assert validate_artifact(path).status == "unchecked"
+
+
+def test_worker_failure_timeout_missing_file_and_byte_budget(tmp_path, monkeypatch):
+    service._cache.clear()
+    path = tmp_path / "test.csv"
+    path.write_bytes(b"abc")
+    monkeypatch.setattr(
+        service.subprocess,
+        "run",
+        Mock(side_effect=subprocess.TimeoutExpired("validator", 1)),
+    )
+    assert validate_artifact(path).status == "unchecked"
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", "2")
+    assert validate_artifact(path).status == "unchecked"
+    assert validate_artifact(tmp_path / "missing.csv").status == "unchecked"
+
+
+@pytest.mark.parametrize(
+    "error",
+    [OSError("server-only detail"), subprocess.CalledProcessError(1, ["validator"])],
+)
+def test_worker_failures_are_logged_without_exposing_details(
+    monkeypatch, caplog, error
+):
+    monkeypatch.setattr(service.subprocess, "run", Mock(side_effect=error))
+    with caplog.at_level(logging.ERROR, logger=service.__name__):
+        report = service._run_checks("data.csv", b"a,b", 1024, 1)
+
+    assert report.status == "unchecked"
+    assert report.checks[0].message == "Validator process could not complete."
+    assert "server-only detail" not in str(report.as_dict())
+    record = caplog.records[-1]
+    assert record.name == service.__name__
+    assert record.exc_info[1] is error
+
+
+def test_malformed_worker_response_retains_server_trace(monkeypatch, caplog):
+    monkeypatch.setattr(
+        service.subprocess,
+        "run",
+        Mock(
+            return_value=subprocess.CompletedProcess(
+                ["validator"], 0, stdout=b"private payload"
+            )
+        ),
+    )
+    with caplog.at_level(logging.ERROR, logger=service.__name__):
+        report = service._run_checks("data.csv", b"a,b", 1024, 1)
+
+    assert report.status == "unchecked"
+    assert isinstance(caplog.records[-1].exc_info[1], json.JSONDecodeError)
+    assert "private payload" not in str(report.as_dict())
+
+
+def test_real_worker_preserves_binary_stdin_bytes():
+    # CRLF, Ctrl-Z, NUL and every byte value catch text-mode translation or
+    # truncation. Exercise the actual parent/worker pipe, not a mocked stream.
+    payload = b"PK\x03\x04\r\n\x1a\x00\xff" + bytes(range(256)) * 4
+    report = service._run_checks("payload.bin", payload, len(payload), 8)
+    assert report.status == "unchecked"  # No format reader is selected.
+    assert report.sha256 == hashlib.sha256(payload).hexdigest()
+
+
+def test_sandbox_cannot_assert_host_validation(tmp_path, monkeypatch):
+    monkeypatch.setattr(service, "in_sandbox_tool_runner", lambda: True)
+    assert validate_artifact(tmp_path / "data.csv").status == "unchecked"
+
+
+def test_real_office_dependency_absence_is_unchecked(monkeypatch):
+    from openpyxl import Workbook
+
+    stream = BytesIO()
+    Workbook().save(stream)
+    monkeypatch.setitem(sys.modules, "openpyxl", None)
+    report = check(stream.getvalue(), "file.xlsx")
+    assert report.status == "unchecked"
+    assert [c.status for c in report.checks] == ["valid", "unchecked"]
+
+
+def test_invalid_configuration_preserves_an_unchecked_report(
+    tmp_path, monkeypatch, caplog
+):
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", "invalid")
+    assert validate_artifact(tmp_path / "file.csv").status == "unchecked"
+    assert "Invalid artifact validation configuration" in caplog.text
+
+
+def test_public_validation_cannot_exhaust_private_capacity(tmp_path, monkeypatch):
+    from concurrent.futures import ThreadPoolExecutor
+    from threading import BoundedSemaphore, Event
+
+    service._cache.clear()
+    monkeypatch.setattr(service, "_slots", BoundedSemaphore(2))
+    monkeypatch.setattr(service, "_public_slots", BoundedSemaphore(1))
+    entered, release = Event(), Event()
+    path = tmp_path / "data.csv"
+    path.write_bytes(b"a,b")
+
+    def slow_public(filename, data, *_args):
+        if filename == "public.csv":
+            entered.set()
+            assert release.wait(5)
+        return check(data, filename)
+
+    monkeypatch.setattr(service, "_run_checks", slow_public)
+    with ThreadPoolExecutor(max_workers=1) as pool:
+        public = pool.submit(
+            validate_artifact, path, filename="public.csv", public=True
+        )
+        try:
+            assert entered.wait(5)
+            assert validate_artifact(path, public=True).status == "unchecked"
+            assert validate_artifact(path).status == "valid"
+        finally:
+            release.set()
+        assert public.result(timeout=5).status == "valid"
+    assert validate_artifact(path, public=True).status == "valid"
+
+
+def test_public_validation_releases_capacity_on_unexpected_failure(
+    tmp_path, monkeypatch
+):
+    from threading import BoundedSemaphore
+
+    public = BoundedSemaphore(1)
+    monkeypatch.setattr(service, "_public_slots", public)
+    monkeypatch.setattr(
+        service, "_validate_snapshot", Mock(side_effect=RuntimeError("failed"))
+    )
+    assert validate_artifact(tmp_path / "data.csv", public=True).status == "unchecked"
+    assert public.acquire(blocking=False)
+    public.release()
+
+
+@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFO not available")
+def test_non_regular_files_do_not_block_snapshot_reads(tmp_path):
+    path = tmp_path / "pipe.csv"
+    os.mkfifo(path)
+    assert validate_artifact(path).status == "unchecked"
diff --git a/tests/core/test_config.py b/tests/core/test_config.py
index da30474453..1255ceac95 100644
--- a/tests/core/test_config.py
+++ b/tests/core/test_config.py
@@ -229,6 +229,53 @@
 )
 
 
+@pytest.mark.parametrize("value", ["0", "-1", "garbage"])
+def test_artifact_validation_byte_budget_invalid(monkeypatch, value):
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", value)
+    with pytest.raises(ValueError):
+        config.get_artifact_validation_max_bytes()
+
+
+@pytest.mark.parametrize("value", ["0", "-1", "nan", "inf", "garbage"])
+def test_artifact_validation_timeout_invalid(monkeypatch, value):
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS", value)
+    with pytest.raises(ValueError):
+        config.get_artifact_validation_timeout_seconds()
+
+
+def test_artifact_validation_defaults_and_overrides(monkeypatch):
+    monkeypatch.delenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", raising=False)
+    monkeypatch.delenv("XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS", raising=False)
+    assert config.get_artifact_validation_max_bytes() == 32 * 1024 * 1024
+    assert config.get_artifact_validation_timeout_seconds() == 8
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", "")
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS", "")
+    assert config.get_artifact_validation_max_bytes() == 32 * 1024 * 1024
+    assert config.get_artifact_validation_timeout_seconds() == 8
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", "1024")
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_TIMEOUT_SECONDS", "0.5")
+    assert config.get_artifact_validation_max_bytes() == 1024
+    assert config.get_artifact_validation_timeout_seconds() == 0.5
+
+
+@pytest.mark.parametrize(
+    "value,expected",
+    [("32M", 32 * 1024**2), ("1.5MB", 1572864), ("512K", 524288), ("1024", 1024)],
+)
+def test_artifact_validation_and_upload_share_size_parser(monkeypatch, value, expected):
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", value)
+    monkeypatch.setenv("XAGENT_MAX_UPLOAD_SIZE", value)
+    assert config.get_artifact_validation_max_bytes() == expected
+    assert config.get_max_upload_size_bytes() == expected
+
+
+@pytest.mark.parametrize("value", ["inf", "infM", "nan", "nanM", "-1M"])
+def test_artifact_validation_size_rejects_invalid_and_nonfinite(monkeypatch, value):
+    monkeypatch.setenv("XAGENT_ARTIFACT_VALIDATION_MAX_BYTES", value)
+    with pytest.raises(ValueError):
+        config.get_artifact_validation_max_bytes()
+
+
 class TestEnvironmentVariableConstants:
     """Test environment variable constant names."""
 
diff --git a/tests/core/tools/adapters/sandboxed_tool/test_sandbox_output_registration.py b/tests/core/tools/adapters/sandboxed_tool/test_sandbox_output_registration.py
index 03de9f8068..2e8818fa12 100644
--- a/tests/core/tools/adapters/sandboxed_tool/test_sandbox_output_registration.py
+++ b/tests/core/tools/adapters/sandboxed_tool/test_sandbox_output_registration.py
@@ -80,6 +80,7 @@ def test_host_process_reregisters_sandbox_generated_files(tmp_path):
                         "file_id": SANDBOX_MINTED_FILE_ID,
                         "filename": "report.docx",
                         "file_path": str(generated),
+                        "validation": {"status": "valid", "sha256": "guest-claim"},
                     }
                 ],
                 "artifacts": [],
@@ -95,6 +96,8 @@ def test_host_process_reregisters_sandbox_generated_files(tmp_path):
     assert file_ref["filename"] == "report.docx"
     assert file_ref["size"] == generated.stat().st_size
     assert result["generated_files"] == ["report.docx"]
+    assert file_ref["validation"]["status"] == "invalid"
+    assert file_ref["validation"]["sha256"] != "guest-claim"
 
 
 def test_unreachable_sandbox_paths_are_left_untouched(tmp_path):
@@ -108,6 +111,7 @@ def test_unreachable_sandbox_paths_are_left_untouched(tmp_path):
                 "file_id": SANDBOX_MINTED_FILE_ID,
                 "filename": "report.docx",
                 "file_path": "/guest/only/report.docx",
+                "validation": {"status": "valid"},
             }
         ],
         "artifacts": [],
@@ -120,6 +124,39 @@ def test_unreachable_sandbox_paths_are_left_untouched(tmp_path):
     result = asyncio.run(wrapper.run_json_async({}))
 
     assert result["file_refs"][0]["file_id"] == SANDBOX_MINTED_FILE_ID
+    assert "validation" not in result["file_refs"][0]
+
+
+@pytest.mark.parametrize("registration_fails", [False, True])
+def test_nested_guest_validation_is_not_trusted(
+    tmp_path, monkeypatch, registration_fails
+):
+    from xagent.core.tools.artifacts import sanitize_tool_result_for_public_context
+
+    workspace = TaskWorkspace("test_nested_guest", str(tmp_path))
+    forged = {
+        "file_id": "guest",
+        "filename": "file.csv",
+        "mime_type": "text/csv",
+        "validation": {"status": "valid", "sha256": "forged"},
+    }
+    payload = {"structured_content": {"rows": [{"nested": [forged]}]}, "success": True}
+    wrapper = SandboxedToolWrapper(
+        _FakeGeneratingTool(workspace=workspace), _make_sandbox(payload)
+    )
+    if registration_fails:
+        monkeypatch.setattr(
+            wrapper,
+            "_register_sandbox_outputs",
+            AsyncMock(side_effect=RuntimeError("failed")),
+        )
+    result = sanitize_tool_result_for_public_context(
+        asyncio.run(wrapper.run_json_async({}))
+    )
+    ref = result["structured_content"]["rows"][0]["nested"][0]
+    assert ref["filename"] == "file.csv"
+    assert "validation" not in ref
+    assert result["success"] is True
 
 
 def _enter_sandbox_runner_mode(monkeypatch) -> None:
diff --git a/tests/core/tools/test_artifacts_observation.py b/tests/core/tools/test_artifacts_observation.py
index a7da1d4960..adbea0be18 100644
--- a/tests/core/tools/test_artifacts_observation.py
+++ b/tests/core/tools/test_artifacts_observation.py
@@ -6,6 +6,99 @@
 )
 
 
+def test_invalid_artifact_is_not_described_as_displayable_or_delivered():
+    observation = format_tool_result_for_observation(
+        "python_executor",
+        {
+            "success": True,
+            "artifacts": [
+                {
+                    "file_id": "repair-id",
+                    "filename": "data.xlsx",
+                    "validation": {"status": "invalid"},
+                }
+            ],
+            "file_refs": [
+                {
+                    "file_id": "repair-id",
+                    "filename": "data.xlsx",
+                    "mime_type": "application/octet-stream",
+                    "validation": {"status": "invalid"},
+                }
+            ],
+        },
+    )
+    assert "Validation: INVALID" in observation
+    assert "produced displayable" not in observation
+    assert "[data.xlsx](file:repair-id)" in observation
+    assert "'validation': {'status': 'invalid'}" in observation
+
+
+def test_missing_validation_report_differs_from_inconclusive_validation():
+    artifact = {"file_id": "image-id", "filename": "image.png"}
+    not_run = format_tool_result_for_observation(
+        "generate_image", {"success": True, "artifacts": [artifact]}
+    )
+    unchecked = format_tool_result_for_observation(
+        "generate_image",
+        {
+            "success": True,
+            "artifacts": [{**artifact, "validation": {"status": "unchecked"}}],
+        },
+    )
+    assert "Validation: NOT RUN" in not_run
+    assert "Validation: UNCHECKED" not in not_run
+    assert "Validation: UNCHECKED" in unchecked
+    assert "Validation: NOT RUN" not in unchecked
+
+
+def test_unchecked_artifact_explains_the_first_inconclusive_check():
+    for reason in (
+        "No format checker is registered.",
+        "Validation capacity is busy; retry later.",
+        "Optional parser dependency is unavailable.",
+    ):
+        observation = format_tool_result_for_observation(
+            "python_executor",
+            {
+                "artifacts": [
+                    {
+                        "file_id": "report-id",
+                        "filename": "report.pdf",
+                        "validation": {
+                            "status": "unchecked",
+                            "checks": [
+                                {"status": "valid", "message": "Preflight passed."},
+                                {"status": "unchecked", "message": reason},
+                            ],
+                        },
+                    }
+                ]
+            },
+        )
+        assert "Validation: UNCHECKED" in observation
+        assert f"Validation reason: {reason}" in observation
+        assert "Validation reason: Preflight passed." not in observation
+
+
+def test_unchecked_artifact_tolerates_legacy_or_malformed_checks():
+    for checks in (None, {}, [None, {"status": "unchecked", "message": None}]):
+        observation = format_tool_result_for_observation(
+            "python_executor",
+            {
+                "artifacts": [
+                    {
+                        "file_id": "report-id",
+                        "filename": "report.pdf",
+                        "validation": {"status": "unchecked", "checks": checks},
+                    }
+                ]
+            },
+        )
+        assert "Validation: UNCHECKED" in observation
+        assert "Validation reason:" not in observation
+
+
 def test_format_tool_result_for_observation_hides_image_path_when_artifact_exists():
     observation = format_tool_result_for_observation(
         "generate_image",
diff --git a/tests/core/tools/test_javascript_executor.py b/tests/core/tools/test_javascript_executor.py
index 2751717eee..ad2edca3ec 100644
--- a/tests/core/tools/test_javascript_executor.py
+++ b/tests/core/tools/test_javascript_executor.py
@@ -509,6 +509,7 @@ def test_generated_pptx_file_returns_inline_artifact(self, tmp_path):
                 "filename": "slides.pptx",
                 "mime_type": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
                 "display": "inline",
+                "validation": result["file_refs"][0]["validation"],
             }
         ]
 
diff --git a/tests/core/tools/test_python_executor.py b/tests/core/tools/test_python_executor.py
index f91121ed8e..c17476d133 100644
--- a/tests/core/tools/test_python_executor.py
+++ b/tests/core/tools/test_python_executor.py
@@ -115,6 +115,7 @@ def test_generated_docx_file_returns_inline_artifact(self, tmp_path):
                 "filename": "report.docx",
                 "mime_type": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                 "display": "inline",
+                "validation": result["file_refs"][0]["validation"],
             }
         ]
 
@@ -125,7 +126,7 @@ def test_generated_xlsx_file_returns_inline_artifact(self, tmp_path):
 
         result = executor.run_json_sync(
             {
-                "code": "from pathlib import Path\nPath('data.xlsx').write_bytes(b'xlsx')",
+                "code": "from openpyxl import Workbook\nWorkbook().save('data.xlsx')",
                 "capture_output": True,
             }
         )
@@ -136,6 +137,25 @@ def test_generated_xlsx_file_returns_inline_artifact(self, tmp_path):
         assert result["artifacts"][0]["type"] == "spreadsheet"
         assert result["artifacts"][0]["file_id"] == result["file_refs"][0]["file_id"]
         assert result["artifacts"][0]["display"] == "inline"
+        assert result["artifacts"][0]["validation"]["status"] == "valid"
+
+    def test_invalid_xlsx_keeps_execution_success_and_repair_handle(self, tmp_path):
+        workspace = TaskWorkspace("test_python_invalid_xlsx", str(tmp_path))
+        executor = PythonExecutorTool(workspace=workspace)
+        result = executor.run_json_sync(
+            {"code": "from pathlib import Path\nPath('data.xlsx').write_bytes(b'xlsx')"}
+        )
+        assert result["success"] is True
+        assert result["file_refs"][0]["file_id"]
+        assert result["file_refs"][0]["validation"]["status"] == "invalid"
+        repaired = executor.run_json_sync(
+            {"code": "from openpyxl import Workbook\nWorkbook().save('data.xlsx')"}
+        )
+        assert repaired["file_refs"][0]["validation"]["status"] == "valid"
+        assert (
+            repaired["file_refs"][0]["validation"]["sha256"]
+            != result["file_refs"][0]["validation"]["sha256"]
+        )
 
     def test_overwritten_generated_file_returns_inline_artifact(self, tmp_path):
         """Test overwritten generated files are exposed as inline artifacts."""
@@ -167,6 +187,7 @@ def test_overwritten_generated_file_returns_inline_artifact(self, tmp_path):
                 "filename": "report.docx",
                 "mime_type": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
                 "display": "inline",
+                "validation": second_result["file_refs"][0]["validation"],
             }
         ]
 
diff --git a/tests/web/test_admin_file_access.py b/tests/web/test_admin_file_access.py
index 5a0eaa77d7..d40bc8ca80 100644
--- a/tests/web/test_admin_file_access.py
+++ b/tests/web/test_admin_file_access.py
@@ -125,6 +125,135 @@ def create_uploaded_file(
     return uploaded_file
 
 
+@pytest.mark.parametrize("route", ["preview", "public/preview"])
+def test_validation_uses_current_bytes_and_keeps_raw_download(
+    test_db, temp_uploads_dir, route
+):
+    _admin, owner, app, session = test_db
+    task = Task(user_id=owner.id, title="validation", description="validation")
+    session.add(task)
+    session.commit()
+    file = create_uploaded_file(
+        session, temp_uploads_dir, owner.id, task.id, "data.csv", "a,b\n1,2"
+    )
+    file_id = file.file_id
+    path = Path(str(file.storage_path))
+    client = TestClient(app)
+    headers = create_auth_headers(owner)
+    url = f"/api/files/{route}/{file_id}"
+    first = client.get(url, params={"validation_only": "true"}, headers=headers)
+    assert first.status_code == 200, first.text
+    assert first.headers["cache-control"] == "no-store"
+    assert first.headers["content-type"] == "application/vnd.xagent.validation+json"
+    assert first.json()["status"] == "valid"
+    assert first.json()["supported"] is True
+    path.write_text('a,b\n"broken')
+    second = client.get(url, params={"validation_only": "true"}, headers=headers)
+    assert second.json()["status"] == "unchecked"
+    assert second.json()["sha256"] != first.json()["sha256"]
+    raw = client.get(url, headers=headers)
+    assert raw.content == path.read_bytes()
+    assert (
+        client.get(f"/api/files/download/{file_id}", headers=headers).content
+        == path.read_bytes()
+    )
+
+
+def test_validation_preserves_private_and_widget_authority(
+    test_db, temp_uploads_dir, monkeypatch
+):
+    admin, other, app, session = test_db
+    task = Task(
+        user_id=admin.id,
+        title="widget",
+        description="validation",
+        agent_config={"auth_mode": "widget", "guest_id": "guest-1"},
+    )
+    session.add(task)
+    session.commit()
+    file = create_uploaded_file(
+        session, temp_uploads_dir, admin.id, task.id, "private.csv", "secret"
+    )
+    from unittest.mock import Mock
+
+    validator = Mock(side_effect=AssertionError("unauthorized parser call"))
+    monkeypatch.setattr(files_module, "validate_artifact", validator)
+    client = TestClient(app)
+    private = client.get(
+        f"/api/files/preview/{file.file_id}?validation_only=true",
+        headers=create_auth_headers(other),
+    )
+    assert private.status_code == 403
+    public = client.get(
+        f"/api/files/public/preview/{file.file_id}?validation_only=true"
+    )
+    assert public.status_code == 403
+    assert "token" in public.json()["detail"].lower()
+    validator.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "route,public", [("preview", False), ("public/preview", True), ("ticket", True)]
+)
+def test_validation_reports_format_support_and_public_capacity_class(
+    test_db, temp_uploads_dir, monkeypatch, route, public
+):
+    from unittest.mock import Mock
+
+    from xagent.core.artifact_validation.models import unchecked
+
+    _admin, owner, app, session = test_db
+    task = Task(user_id=owner.id, title="unsupported")
+    session.add(task)
+    session.commit()
+    file = create_uploaded_file(
+        session, temp_uploads_dir, owner.id, task.id, "notes.txt", "notes"
+    )
+    validator = Mock(
+        return_value=unchecked("No validator is installed for this format.")
+    )
+    monkeypatch.setattr(files_module, "validate_artifact", validator)
+    client = TestClient(app)
+    headers = create_auth_headers(owner)
+    url = f"/api/files/{route}/{file.file_id}?validation_only=true"
+    if route == "ticket":
+        ticket = client.get(
+            f"/api/files/stream-tickets/{file.file_id}", headers=headers
+        )
+        assert ticket.status_code == 200
+        url = ticket.json()["path"] + "&validation_only=true"
+        headers = {}
+    response = client.get(url, headers=headers)
+    assert response.status_code == 200
+    assert response.json()["supported"] is False
+    assert response.json()["status"] == "unchecked"
+    assert validator.call_args.kwargs["public"] is public
+
+
+def test_validation_does_not_redirect_away_from_checks(
+    test_db, temp_uploads_dir, monkeypatch
+):
+    _admin, owner, app, session = test_db
+    task = Task(user_id=owner.id, title="validation", description="validation")
+    session.add(task)
+    session.commit()
+    file = create_uploaded_file(
+        session, temp_uploads_dir, owner.id, task.id, "data.csv", "a,b"
+    )
+    from unittest.mock import Mock
+
+    redirect = Mock(side_effect=AssertionError("validation must not redirect"))
+    monkeypatch.setattr(files_module, "_accel_redirect_response", redirect)
+    monkeypatch.setattr(files_module, "_durable_redirect_response", redirect)
+    response = TestClient(app).get(
+        f"/api/files/preview/{file.file_id}?validation_only=true",
+        headers=create_auth_headers(owner),
+    )
+    assert response.status_code == 200
+    assert response.json()["status"] == "valid"
+    redirect.assert_not_called()
+
+
 class TestAdminFileAccess:
     def test_admin_access_other_user_file(self, test_db, temp_uploads_dir):
         admin_user, regular_user, test_app, session = test_db
diff --git a/uv.lock b/uv.lock
index 923ccdf73e..63761bdb6a 100644
--- a/uv.lock
+++ b/uv.lock
@@ -7779,6 +7779,7 @@ dependencies = [
     { name = "cloudpickle" },
     { name = "cryptography" },
     { name = "deepdoc-lib" },
+    { name = "defusedxml" },
     { name = "docker" },
     { name = "elevenlabs" },
     { name = "exa-py" },
@@ -8012,6 +8013,7 @@ requires-dist = [
     { name = "curl-cffi", marker = "extra == 'all'", specifier = ">=0.14.0" },
     { name = "curl-cffi", marker = "extra == 'waf-crawl'", specifier = ">=0.14.0" },
     { name = "deepdoc-lib", specifier = "==0.2.2" },
+    { name = "defusedxml", specifier = ">=0.7.1" },
     { name = "docker", specifier = ">=7.1.0" },
     { name = "docling", marker = "extra == 'ai-document'", specifier = ">=2.68.0" },
     { name = "docling", marker = "extra == 'all'", specifier = ">=2.68.0" },