Forward-merge release/2.2 into develop

ci/markdown-link-check-config.json

@@ -35,6 +35,9 @@
         },
         {
             "pattern": "^https?://huggingface\\.co/spaces/muset-ai/DeepResearch-Bench-Leaderboard$"
+        },
+        {
+            "pattern": "^https://you\\.com/docs/api-reference/contents/?$"
         }
     ],
     "replacementPatterns": [

docs/source/examples/skills-sandbox/index.md

@@ -107,9 +107,8 @@ functions:
 AI-Q validates the public skill collection names (`research`, `synthesis`) and resolves them to DeepAgents source paths internally. When skills are configured, AI-Q mounts the configured built-in skill collections into the DeepAgents virtual filesystem. When the sandbox ref is present, DeepAgents `execute` calls run in the configured provider. Modal creates a fresh sandbox named for the job.
 
 In the reference async API flow, artifact capture uses the job database configured by
-`general.front_end.db_url` (`NAT_JOB_STORE_DB_URL`) for metadata. Without that job-scoped database URL, as in a direct
-`nat run`, Modal execution still works but durable capture remains inactive. Artifact bytes use SQL BLOB storage in the
-job database by default. For production, use S3-compatible object storage by setting `AIQ_ARTIFACT_BLOB_PROVIDER=s3`,
+`general.front_end.db_url` (`NAT_JOB_STORE_DB_URL`) for metadata. Artifact bytes use SQL BLOB storage in the job database
+by default. For production, use S3-compatible object storage by setting `AIQ_ARTIFACT_BLOB_PROVIDER=s3`,
 `AIQ_ARTIFACT_S3_BUCKET`, and the standard AWS credentials; set `AIQ_ARTIFACT_S3_ENDPOINT_URL` for MinIO or another
 compatible service. See [Production Artifact Storage](../../deployment/production.md#artifact-storage) for all options.
 
@@ -120,15 +119,32 @@ deletes the sandbox at terminal cleanup. Attaching to an existing shared
 sandbox is available only through explicit debug settings and is not
 job-isolated.
 
-## Run AI-Q
+## Run Synchronously with `nat run` (Non-Persistent)
+
+```{warning}
+`nat run` is a synchronous, single-run command. It does not create an async job record or connect the workflow to the
+job-scoped artifact store. The final report is returned normally, and `/shared/` files can contribute to that report
+during the run, but run state, `/shared/` content, and sandbox-generated files cannot be retrieved through the job or
+artifact APIs after the command finishes.
+```
 
 ```bash
 dotenv -f deploy/.env run .venv/bin/nat run \
   --config_file configs/config_domain_routing_and_skills.yml \
   --input "Compare the top 10 publicly traded semiconductor companies by 2024 revenue. Build a markdown table with revenue, YoY growth, market cap, and gross margin. Then rank them and compute summary statistics. Use the data analysis tool for all calculations."
 ```
 
-For API or UI testing:
+Use this mode to try the workflow when you only need its returned report. Do not use it when you need durable job state
+or separately retrievable charts, CSVs, notebooks, or other generated files.
+
+## Run with `nat serve` for Persistent Jobs and Artifacts
+
+To retain job information and retrieve supported generated files after a run, start the async API with `nat serve`.
+The configured `NAT_JOB_STORE_DB_URL` supplies the required job-scoped store. The reference config defaults to a local
+SQLite database; production deployments should configure PostgreSQL and appropriate artifact blob storage.
+For trusted local development, set `REQUIRE_AUTH=false` in `deploy/.env`; the commands below omit credentials on that
+basis. When `REQUIRE_AUTH=true`, these job routes require authentication, so configure authentication and add the same
+`Authorization: Bearer $AIQ_TOKEN` header to every `curl` command below.
 
 ```bash
 dotenv -f deploy/.env run .venv/bin/nat serve \
@@ -137,7 +153,37 @@ dotenv -f deploy/.env run .venv/bin/nat serve \
   --port 8000
 ```
 
-Then submit a deep research request through the AI-Q API or UI.
+In another terminal, submit a deep research request with a known job ID. Custom job IDs must be unique, so change this
+value before repeating the example against the same job store:
+
+```bash
+curl -X POST http://localhost:8000/v1/jobs/async/submit \
+  -H "Content-Type: application/json" \
+  -d '{
+    "job_id": "skills-sandbox-example",
+    "agent_type": "deep_researcher",
+    "input": "Compare the top 10 publicly traded semiconductor companies by 2024 revenue. Build a markdown table and a CSV with revenue, YoY growth, market cap, and gross margin. Then rank them and compute summary statistics. Use the data analysis tool for all calculations."
+  }'
+```
+
+Check the job until its status is either `success` or `failure`. If it reaches `failure`, inspect the response's `error`
+field for the actionable failure message:
+
+```bash
+curl http://localhost:8000/v1/jobs/async/job/skills-sandbox-example
+```
+
+List its captured artifacts, then use an `artifact_id` from the response to download one:
+
+```bash
+curl http://localhost:8000/v1/jobs/async/job/skills-sandbox-example/artifacts
+curl -OJ http://localhost:8000/v1/jobs/async/job/skills-sandbox-example/artifacts/{artifact_id}/content
+```
+
+Artifact capture is best-effort and limited to the configured file types and size. Stored artifacts are retrievable
+rather than permanent: server-wide retention cleanup can remove an artifact independently of the job's expiry. For the
+complete API contract, authentication guidance, and retention behavior, see
+[Durable Sandbox Artifacts](../../integration/rest-api.md#durable-sandbox-artifacts).
 
 ## Example Queries
 
@@ -230,8 +276,7 @@ No config change is required for additional built-in skills inside an enabled co
 - Text artifacts that need to survive for the report should be written through DeepAgents filesystem tools to `/shared/...`.
 - `/shared/` is a virtual DeepAgents filesystem path. Use `ls`, `read_file`, `write_file`, and `edit_file` for `/shared/`; do not inspect `/shared/` with shell commands through `execute`.
 - The sandbox is configured with `network: blocked`, so research should happen through AI-Q search tools, not from sandbox code.
-- The reference profile enables durable sandbox artifact capture, which requires the async API's job-scoped artifact
-  store. Successful `execute` calls checkpoint manifest-declared files, and success/failure terminal paths perform one
-  final best-effort scan. A busy cancellation skips that scan and preserves earlier checkpoints. Direct `nat run` does
-  not provide that store, and adding a sandbox alone does not guarantee that every generated file is persisted or
-  embedded in the report.
+- The reference profile enables durable sandbox artifact capture for async API jobs. Successful `execute` calls
+  checkpoint manifest-declared files, and success/failure terminal paths perform one final best-effort scan. A busy
+  cancellation skips that scan and preserves earlier checkpoints. Adding a sandbox alone does not guarantee that every
+  generated file is persisted or embedded in the report.

docs/source/integration/rest-api.md

@@ -204,6 +204,13 @@ curl http://localhost:8000/v1/jobs/async/job/{job_id}
 
 Job statuses: `SUBMITTED`, `RUNNING`, `SUCCESS`, `FAILURE`, `INTERRUPTED`.
 
+Typed source-condition failures, such as running with no selected sources or receiving no
+results from the selected sources, remain in `FAILURE`. Their `error` is an actionable
+message describing how to correct the source selection or query. If the agent produced a
+sanitized answer before detecting the source condition, that report remains available from
+`GET /v1/jobs/async/job/{job_id}/report`. Unexpected failures continue to return a sanitized
+error and do not expose internal exception details or plaintext output.
+
 ### Stream Events (SSE)
 
 Stream real-time events from a running or completed job using Server-Sent Events.

frontends/aiq_api/src/aiq_api/jobs/callbacks.py

@@ -126,6 +126,15 @@ def to_sse_dict(self) -> dict:
         return {k: v for k, v in result.items() if v is not None}
 
 
+def build_final_report_event(content: str) -> IntermediateStepEvent:
+    """Build the canonical final-report artifact event."""
+    return IntermediateStepEvent(
+        category=EventCategory.ARTIFACT,
+        state=EventState.UPDATE,
+        data=EventData(type=ArtifactType.OUTPUT.value, content=content, output_category="final_report"),
+    )
+
+
 class ToolArtifactMapping:
     """
     Maps tool names to artifact types for automatic artifact emission.
@@ -381,11 +390,7 @@ def emit_final_report(self, content: str) -> None:
         frontend receives the verified content (overwrites the earlier
         auto-emitted version).
         """
-        self._emit_artifact(
-            ArtifactType.OUTPUT,
-            content,
-            output_category="final_report",
-        )
+        self._emit(build_final_report_event(content))
 
     def _is_search_tool(self, tool_name: str) -> bool:
         """Check if tool is a search-related tool that returns URLs."""

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

@@ -35,7 +35,10 @@
 from typing import TYPE_CHECKING
 from typing import Any
 
+from aiq_agent.common.citation_verification import EmptySourceRegistryError
+
 from .callbacks import AgentEventCallback
+from .callbacks import build_final_report_event
 from .event_store import BatchingEventStore
 from .event_store import EventStore
 
@@ -238,6 +241,109 @@ def _write_job_success_if_running_sync(db_url: str, job_id: str, stored_output:
         return (result.rowcount or 0) == 1
 
 
+def _write_job_source_failure_if_running_sync(
+    db_url: str,
+    job_id: str,
+    public_error: str,
+    stored_output: str,
+    final_report_event: dict[str, Any] | None = None,
+    job_output_cipher: Any | None = None,
+) -> bool:
+    """Persist a source failure, then store its optional final-report event best-effort."""
+    from sqlalchemy import text
+
+    from .event_store import EventStore
+
+    engine = EventStore._get_or_create_sync_engine(db_url)
+    stmt = text(
+        f"UPDATE job_info SET status = 'failure', error = :error, output = :output, "
+        f"updated_at = {_db_now_expr(db_url)} WHERE job_id = :job_id AND status = 'running'"
+    )
+    with engine.begin() as conn:
+        result = conn.execute(
+            stmt,
+            {"error": public_error, "output": stored_output, "job_id": job_id},
+        )
+        if (result.rowcount or 0) != 1:
+            return False
+
+    if final_report_event is not None:
+        try:
+            EventStore(db_url, job_id, content_cipher=job_output_cipher).store(final_report_event)
+        except Exception as exc:
+            logger.warning(
+                "Job %s source-failure final-report event write failed exception=%s",
+                job_id,
+                exc.__class__.__name__,
+            )
+    return True
+
+
+def _write_job_failure_if_running_sync(db_url: str, job_id: str, public_error: str) -> bool:
+    """Compare-and-set a generic failure without changing an existing terminal job."""
+    from sqlalchemy import text
+
+    from .event_store import EventStore
+
+    engine = EventStore._get_or_create_sync_engine(db_url)
+    stmt = text(
+        f"UPDATE job_info SET status = 'failure', error = :error, updated_at = {_db_now_expr(db_url)} "
+        "WHERE job_id = :job_id AND status = 'running'"
+    )
+    with engine.begin() as conn:
+        result = conn.execute(stmt, {"error": public_error, "job_id": job_id})
+        return (result.rowcount or 0) == 1
+
+
+async def _persist_empty_source_failure(
+    *,
+    error: EmptySourceRegistryError,
+    job_output_cipher: Any,
+    db_url: str,
+    job_id: str,
+    event_store: Any | None = None,
+) -> bool:
+    """Persist a typed source failure, falling back safely if output storage fails."""
+    from .crypto import serialize_job_output_for_storage
+
+    output = {
+        "report": error.generated_answer,
+        "outcome_reason": error.reason.value,
+    }
+    try:
+        stored_output = serialize_job_output_for_storage(output, job_output_cipher)
+        if event_store is not None and hasattr(event_store, "flush"):
+            await asyncio.to_thread(event_store.flush)
+        final_report_event = (
+            build_final_report_event(error.generated_answer).to_sse_dict() if error.generated_answer else None
+        )
+        return await asyncio.get_running_loop().run_in_executor(
+            None,
+            _write_job_source_failure_if_running_sync,
+            db_url,
+            job_id,
+            error.public_message,
+            stored_output,
+            final_report_event,
+            job_output_cipher,
+        )
+    except Exception as exc:
+        logger.warning(
+            "Job %s source-failure output write failed exception=%s",
+            job_id,
+            exc.__class__.__name__,
+        )
+        sanitized_error = f"job failed ({type(exc).__name__}); check server logs for details"
+        await asyncio.get_running_loop().run_in_executor(
+            None,
+            _write_job_failure_if_running_sync,
+            db_url,
+            job_id,
+            sanitized_error,
+        )
+        return False
+
+
 def _run_lease_refresher(db_url: str, job_id: str, stop_event: threading.Event) -> None:
     """Refresh the running-job lease on a dedicated thread until signalled.
 
@@ -958,6 +1064,24 @@ async def run_agent_job(
             except (ConnectionError, TimeoutError, RuntimeError):
                 pass
 
+    except EmptySourceRegistryError as e:
+        logger.info("Job %s failed because no research sources were available (%s)", job_id, e.reason.value)
+        if event_store is None:
+            event_store = BatchingEventStore(EventStore(db_url, job_id, content_cipher=job_output_cipher))
+
+        await asyncio.to_thread(_harvest_sandbox_artifacts, sandbox_runtime, job_id=job_id, interrupted=False)
+        wrote = await _persist_empty_source_failure(
+            error=e,
+            job_output_cipher=job_output_cipher,
+            db_url=db_url,
+            job_id=job_id,
+            event_store=event_store,
+        )
+        if wrote:
+            logger.info("Job %s persisted source-failure outcome", job_id)
+        else:
+            logger.warning("Job %s already terminal or source-failure output persistence failed", job_id)
+
     except Exception as e:
         logger.exception("Job %s failed: %s", job_id, type(e).__name__)
         if event_store is None:

frontends/aiq_api/tests/test_content_encryption_routes.py

@@ -80,7 +80,15 @@ def _enable_vault(monkeypatch) -> None:
     crypto.reset_content_encryption_manager_for_tests()
 
 
-async def _build_jobs_app(monkeypatch, tmp_path, *, job_output=None, submitted_job=None) -> FastAPI:
+async def _build_jobs_app(
+    monkeypatch,
+    tmp_path,
+    *,
+    job_output=None,
+    job_status="success",
+    job_error=None,
+    submitted_job=None,
+) -> FastAPI:
     import aiq_api.routes.jobs as jobs_routes
     from aiq_api.jobs import access
     from aiq_api.jobs import event_store
@@ -112,8 +120,8 @@ async def _no_op_reaper(*_args, **_kwargs):
 
     job = SimpleNamespace(
         job_id="job-1",
-        status="success",
-        error=None,
+        status=job_status,
+        error=job_error,
         output=job_output,
         created_at=datetime.now(UTC),
     )
@@ -583,6 +591,53 @@ async def test_report_decrypts_encrypted_final_output(monkeypatch, tmp_path):
     }
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("outcome_reason", "actionable_error"),
+    [
+        (
+            "no_sources_selected",
+            "No data sources are selected. Select at least one data source and run the research again.",
+        ),
+        (
+            "no_source_results",
+            "The selected data sources returned no results. "
+            "Try rephrasing the question or selecting different data sources.",
+        ),
+    ],
+)
+async def test_source_failure_status_is_actionable_and_preserved_report_is_available(
+    monkeypatch, tmp_path, outcome_reason, actionable_error
+):
+    _enable_static_key(monkeypatch)
+    stored = crypto.create_job_content_cipher("job-1").encrypt_output_json(
+        json.dumps(
+            {
+                "report": "# Preserved generated answer",
+                "outcome_reason": outcome_reason,
+            }
+        )
+    )
+    app = await _build_jobs_app(
+        monkeypatch,
+        tmp_path,
+        job_output=stored,
+        job_status="failure",
+        job_error=actionable_error,
+    )
+
+    with TestClient(app) as client:
+        status_response = client.get("/v1/jobs/async/job/job-1")
+        report_response = client.get("/v1/jobs/async/job/job-1/report")
+
+    assert status_response.status_code == 200
+    assert status_response.json()["status"] == "failure"
+    assert status_response.json()["error"] == actionable_error
+    assert report_response.status_code == 200
+    assert report_response.json()["has_report"] is True
+    assert report_response.json()["report"] == "# Preserved generated answer"
+
+
 @pytest.mark.asyncio
 async def test_state_returns_500_for_invalid_encrypted_event_data(monkeypatch, tmp_path):
     from aiq_api.jobs.event_store import EventStore