feat: persist and surface sandbox artifacts

docs/source/architecture/agents/sandbox.md

@@ -51,6 +51,9 @@ and [Production Considerations](../../deployment/production.md#artifact-storage)
   caps (`AIQ_MAX_SANDBOXES_PER_PRINCIPAL` / `AIQ_MAX_SANDBOXES_GLOBAL`, default-off) bound
   concurrency/cost but do not provide filesystem isolation.
 - Custom client-supplied job IDs must not be reused for a new job.
+- Manifest checkpoints preserve completed artifacts after successful sandbox commands. The
+  terminal finalizer harvests once before cleanup on success/failure; cancellation harvests
+  only when the provider is idle and otherwise terminates immediately.
 - The runtime closes provider sessions on success, failure, cancellation, and timeout.
   A named OpenShell sandbox persists when `delete_on_exit` is disabled.
 

frontends/aiq_api/README.md

@@ -149,7 +149,7 @@ Events streamed during job execution:
 | `workflow.start` / `workflow.end` | Workflow lifecycle |
 | `llm.start` / `llm.chunk` / `llm.end` | LLM inference progress |
 | `tool.start` / `tool.end` | Tool invocations |
-| `artifact.update` | Todos, files, citations, output updates |
+| `artifact.update` | Todos, citations, output, and generated-file metadata (`content_url` for durable bytes) |
 | `job.error` | Error occurred |
 
 ## Configuration

frontends/aiq_api/src/aiq_api/jobs/runner.py

@@ -798,7 +798,16 @@ async def run_agent_job(
                     # Signal event stream completion
                     event_stream.on_complete()
 
-                    # Flush any buffered events before updating status
+                    # Harvest artifacts (durable, idempotent) before SUCCESS so clients cannot
+                    # stop streaming before the terminal metadata is persisted. Resource release
+                    # is deferred to the finally block: the provider's close() is unbounded, so
+                    # awaiting it here could strand a finished job in RUNNING if SDK cleanup hangs.
+                    await asyncio.to_thread(
+                        _harvest_sandbox_artifacts,
+                        sandbox_runtime,
+                        job_id=job_id,
+                        interrupted=False,
+                    )
                     if hasattr(event_store, "flush"):
                         event_store.flush()
 
@@ -832,17 +841,10 @@ async def run_agent_job(
     except asyncio.CancelledError:
         logger.info("Job %s cancelled", job_id)
         interrupted = True
-        if job_store:
-            try:
-                job = await job_store.get_job(job_id)
-                if job and job.status != JobStatus.INTERRUPTED.value:
-                    await job_store.update_status(job_id, JobStatus.INTERRUPTED, error="cancelled by user")
-            except (ConnectionError, TimeoutError, RuntimeError):
-                pass
-
         if event_store is None:
             event_store = BatchingEventStore(EventStore(db_url, job_id))
 
+        await asyncio.to_thread(_teardown_sandbox, sandbox_runtime, job_id=job_id, interrupted=True)
         _store_terminal_event_best_effort(
             event_store,
             {
@@ -851,14 +853,20 @@ async def run_agent_job(
             },
         )
 
-    except Exception as e:
-        logger.exception("Job %s failed: %s", job_id, type(e).__name__)
         if job_store:
-            await job_store.update_status(job_id, JobStatus.FAILURE, error=str(e))
+            try:
+                job = await job_store.get_job(job_id)
+                if job and job.status != JobStatus.INTERRUPTED.value:
+                    await job_store.update_status(job_id, JobStatus.INTERRUPTED, error="cancelled by user")
+            except (ConnectionError, TimeoutError, RuntimeError):
+                pass
 
+    except Exception as e:
+        logger.exception("Job %s failed: %s", job_id, type(e).__name__)
         if event_store is None:
             event_store = BatchingEventStore(EventStore(db_url, job_id))
 
+        await asyncio.to_thread(_harvest_sandbox_artifacts, sandbox_runtime, job_id=job_id, interrupted=False)
         _store_terminal_event_best_effort(
             event_store,
             {
@@ -869,6 +877,8 @@ async def run_agent_job(
                 },
             },
         )
+        if job_store:
+            await job_store.update_status(job_id, JobStatus.FAILURE, error=str(e))
 
     finally:
         # Ensure terminal-path events are not left in the batch buffer.
@@ -883,9 +893,7 @@ async def run_agent_job(
                 )
         if cancellation_monitor:
             cancellation_monitor.stop()
-        # Release the sandbox off the event loop so the SDK session close never blocks the Dask
-        # worker. The single artifact harvest already ran in agent.run() before this point, so
-        # teardown only closes/terminates; interrupted jobs terminate() to preempt a live execute.
+        # Idempotent fallback for failures before a terminal branch finalized the runtime.
         await asyncio.to_thread(_teardown_sandbox, sandbox_runtime, job_id=job_id, interrupted=interrupted)
         # Clean up job-scoped auth token
         if _auth_token_reset is not None:
@@ -909,24 +917,50 @@ def _store_terminal_event_best_effort(event_store, event: dict) -> None:
         )
 
 
+def _harvest_sandbox_artifacts(sandbox_runtime: Any | None, *, job_id: str, interrupted: bool) -> None:
+    """Persist captured artifacts on a terminal path without releasing the sandbox.
+
+    Callable before the terminal job status so artifact metadata is durable, yet it never invokes
+    the provider's unbounded ``close()``/``terminate()``. Resource release stays in
+    ``_teardown_sandbox`` (run from ``finally``) so a hanging SDK cleanup cannot strand a finished
+    job in ``RUNNING`` with a stream that never terminates. The harvest is idempotent.
+    """
+    if sandbox_runtime is None:
+        return
+    finalize_artifacts = getattr(sandbox_runtime, "finalize_artifacts", None)
+    if callable(finalize_artifacts):
+        try:
+            finalize_artifacts(interrupted=interrupted)
+        except Exception as exc:  # noqa: BLE001 - artifact capture cannot replace the job result
+            logger.warning(
+                "Terminal artifact harvest failed for job %s exception=%s",
+                job_id,
+                exc.__class__.__name__,
+            )
+
+
 def _teardown_sandbox(sandbox_runtime: Any | None, *, job_id: str, interrupted: bool) -> None:
-    """Release sandbox resources on a terminal path (best-effort, never raises).
+    """Harvest artifacts and release sandbox resources on a terminal path.
 
     Interrupted jobs (cancel/timeout) call ``terminate()`` so a still-running ``execute`` is
     forcibly preempted; normal paths call ``close()`` gracefully. Both are idempotent. This runs
     off the event loop (``asyncio.to_thread``) so the SDK session close cannot block the worker.
     """
     if sandbox_runtime is None:
         return
+    _harvest_sandbox_artifacts(sandbox_runtime, job_id=job_id, interrupted=interrupted)
     teardown = getattr(sandbox_runtime, "terminate", None) if interrupted else None
     if teardown is None:
         teardown = getattr(sandbox_runtime, "close", None)
     if teardown is None:
         return
     try:
         teardown()
-    except Exception:  # noqa: BLE001 - cleanup must never raise on the terminal path
-        logger.warning("Sandbox cleanup failed for job %s", job_id, exc_info=True)
+    except Exception as exc:  # noqa: BLE001 - cleanup must never raise on the terminal path
+        # Secret-safe: log only the exception type. A provider cleanup error can carry a
+        # credential or internal hostname, which must never reach the logs (matches the
+        # finalize_artifacts handler above).
+        logger.warning("Sandbox cleanup failed for job %s exception=%s", job_id, exc.__class__.__name__)
 
 
 def _create_agent_instance(

frontends/ui/src/adapters/api/deep-research-client.spec.ts

@@ -2,7 +2,29 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { afterEach, describe, expect, test, vi } from 'vitest'
-import { getJobStatus } from './deep-research-client'
+import { createDeepResearchClient, getJobStatus } from './deep-research-client'
+
+class FakeEventSource {
+  static latest: FakeEventSource | null = null
+  onopen: (() => void) | null = null
+  onmessage: ((event: MessageEvent) => void) | null = null
+  onerror: (() => void) | null = null
+  private listeners = new Map<string, (event: MessageEvent) => void>()
+
+  constructor(_url: string) {
+    FakeEventSource.latest = this
+  }
+
+  addEventListener(type: string, listener: EventListener): void {
+    this.listeners.set(type, listener as (event: MessageEvent) => void)
+  }
+
+  close(): void {}
+
+  emit(type: string, data: unknown): void {
+    this.listeners.get(type)?.(new MessageEvent(type, { data: JSON.stringify(data) }))
+  }
+}
 
 describe('deep research REST client', () => {
   afterEach(() => {
@@ -32,4 +54,82 @@ describe('deep research REST client', () => {
       'Failed to get job status: 500 - PROXY_ERROR: fetch failed'
     )
   })
+
+  test('maps generated binary file events to durable artifact metadata', () => {
+    vi.stubGlobal('EventSource', FakeEventSource)
+    const onFileUpdate = vi.fn()
+    const client = createDeepResearchClient({ jobId: 'job-1', callbacks: { onFileUpdate } })
+    client.connect()
+
+    FakeEventSource.latest?.emit('artifact.update', {
+      data: {
+        type: 'file',
+        file_path: 'chart.png',
+        artifact_id: 'art_123',
+        content_url: '/v1/jobs/async/job/job-1/artifacts/art_123/content',
+        kind: 'image',
+        mime_type: 'image/png',
+        size_bytes: 2048,
+        sha256: 'a'.repeat(64),
+        title: 'Quarterly CapEx',
+        caption: 'Comparison chart',
+        inline: true,
+      },
+    })
+
+    expect(onFileUpdate).toHaveBeenCalledWith({
+      filename: 'chart.png',
+      content: undefined,
+      artifactId: 'art_123',
+      contentUrl: '/api/jobs/async/job/job-1/artifacts/art_123/content',
+      kind: 'image',
+      mimeType: 'image/png',
+      sizeBytes: 2048,
+      sha256: 'a'.repeat(64),
+      title: 'Quarterly CapEx',
+      caption: 'Comparison chart',
+      inline: true,
+    })
+  })
+
+  test('preserves legacy text file events', () => {
+    vi.stubGlobal('EventSource', FakeEventSource)
+    const onFileUpdate = vi.fn()
+    const client = createDeepResearchClient({ jobId: 'job-1', callbacks: { onFileUpdate } })
+    client.connect()
+
+    FakeEventSource.latest?.emit('artifact.update', {
+      data: { type: 'file', file_path: 'report.md', content: '# Report' },
+    })
+
+    expect(onFileUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({ filename: 'report.md', content: '# Report' })
+    )
+  })
+
+  test('maps url-only artifacts (no artifact_id) via the content_url/url fallback', () => {
+    vi.stubGlobal('EventSource', FakeEventSource)
+    const onFileUpdate = vi.fn()
+    const client = createDeepResearchClient({ jobId: 'job-1', callbacks: { onFileUpdate } })
+    client.connect()
+
+    // No artifact_id: filename derives from the url and contentUrl falls back to content_url.
+    FakeEventSource.latest?.emit('artifact.update', {
+      data: {
+        type: 'file',
+        url: 'https://example.com/exports/data.csv',
+        content_url: 'https://example.com/exports/data.csv',
+        mime_type: 'text/csv',
+      },
+    })
+
+    expect(onFileUpdate).toHaveBeenCalledWith(
+      expect.objectContaining({
+        filename: 'data.csv',
+        artifactId: undefined,
+        contentUrl: 'https://example.com/exports/data.csv',
+        mimeType: 'text/csv',
+      })
+    )
+  })
 })

frontends/ui/src/adapters/api/deep-research-client.ts

@@ -12,6 +12,7 @@
  */
 
 import { apiConfig } from './config'
+import { artifactContentPath } from '@/shared/utils/artifact-url'
 
 // ============================================================
 // Types
@@ -149,6 +150,21 @@ export interface TodoItem {
   status: 'pending' | 'in_progress' | 'completed' | 'cancelled'
 }
 
+/** File update emitted by artifact.update. Durable generated files carry metadata, not bytes. */
+export interface FileArtifactUpdate {
+  filename: string
+  content?: string
+  artifactId?: string
+  contentUrl?: string
+  kind?: string
+  mimeType?: string
+  sizeBytes?: number
+  sha256?: string
+  title?: string
+  caption?: string
+  inline?: boolean
+}
+
 /** artifact.update event */
 export interface ArtifactUpdateEvent extends DeepResearchSSEEvent {
   event: 'artifact.update'
@@ -157,8 +173,19 @@ export interface ArtifactUpdateEvent extends DeepResearchSSEEvent {
     timestamp: string
     data: {
       type: ArtifactType
-      content: string | TodoItem[]
+      content?: string | TodoItem[]
       url?: string // For citation_source and citation_use types
+      content_url?: string
+      file_path?: string
+      artifact_id?: string
+      job_id?: string
+      kind?: string
+      mime_type?: string
+      size_bytes?: number
+      sha256?: string
+      title?: string
+      caption?: string
+      inline?: boolean
     }
     metadata?: {
       workflow?: string
@@ -208,7 +235,7 @@ export interface DeepResearchCallbacks {
   /** Called on artifact updates */
   onTodoUpdate?: (todos: TodoItem[], workflow?: string) => void
   onCitationUpdate?: (url: string, content: string, isCited?: boolean) => void
-  onFileUpdate?: (filename: string, content: string) => void
+  onFileUpdate?: (file: FileArtifactUpdate) => void
   onOutputUpdate?: (content: string, outputCategory?: string, workflow?: string) => void
   /** Called on job heartbeat (confirms job is alive during long operations) */
   onHeartbeat?: (uptimeSeconds: number) => void
@@ -485,10 +512,17 @@ export const createDeepResearchClient = (options: DeepResearchStreamOptions): De
       case 'artifact.update': {
         // artifact.update has nested structure: { id, timestamp, data: { type, content, url?, output_category? }, metadata?: { workflow } }
         const artifactWrapper = rawData as {
-          data?: { type: ArtifactType; content: string | TodoItem[]; url?: string; output_category?: string }
+          data?: {
+            type: ArtifactType
+            content?: string | TodoItem[]
+            url?: string
+            content_url?: string
+            output_category?: string
+          }
           type?: ArtifactType
           content?: string | TodoItem[]
           url?: string
+          content_url?: string
           output_category?: string
           metadata?: { workflow?: string }
         }
@@ -509,11 +543,32 @@ export const createDeepResearchClient = (options: DeepResearchStreamOptions): De
             callbacks.onCitationUpdate?.(artifactData.url || '', artifactData.content as string, true)
             break
           case 'file': {
-            // file artifacts are written during research — extract filename from path
+            // Generated artifacts carry durable metadata; legacy text-file events carry content.
             const raw = artifactData as Record<string, unknown>
-            const filePath = (raw.file_path || raw.path || artifactData.url || 'unknown') as string
-            const fileName = filePath.split('/').pop() || filePath
-            callbacks.onFileUpdate?.(fileName, artifactData.content as string)
+            const artifactId = typeof raw.artifact_id === 'string' ? raw.artifact_id : undefined
+            // Fall back to artifactId (not a shared 'unknown') when no path/url is present, so two
+            // distinct pathless artifacts don't collapse onto one filename key downstream.
+            const filePath = (raw.file_path || raw.path || artifactData.url) as string | undefined
+            const fileName = filePath ? filePath.split('/').pop() || filePath : artifactId || 'unknown'
+            callbacks.onFileUpdate?.({
+              filename: fileName,
+              content: typeof artifactData.content === 'string' ? artifactData.content : undefined,
+              artifactId,
+              contentUrl: artifactId
+                ? artifactContentPath(jobId, artifactId)
+                : typeof raw.content_url === 'string'
+                  ? raw.content_url
+                  : typeof raw.url === 'string'
+                    ? raw.url
+                    : undefined,
+              kind: typeof raw.kind === 'string' ? raw.kind : undefined,
+              mimeType: typeof raw.mime_type === 'string' ? raw.mime_type : undefined,
+              sizeBytes: typeof raw.size_bytes === 'number' ? raw.size_bytes : undefined,
+              sha256: typeof raw.sha256 === 'string' ? raw.sha256 : undefined,
+              title: typeof raw.title === 'string' ? raw.title : undefined,
+              caption: typeof raw.caption === 'string' ? raw.caption : undefined,
+              inline: typeof raw.inline === 'boolean' ? raw.inline : undefined,
+            })
             break
           }
           case 'output':

frontends/ui/src/adapters/api/index.ts

@@ -122,6 +122,7 @@ export type {
   ToolStartEvent,
   ToolEndEvent,
   TodoItem,
+  FileArtifactUpdate,
   ArtifactUpdateEvent,
   DeepResearchEvent,
   DeepResearchCallbacks,