diff --git a/deploy/.env.example b/deploy/.env.example index 472f0f49f..319d8ae4c 100644 --- a/deploy/.env.example +++ b/deploy/.env.example @@ -47,6 +47,21 @@ DASK_DISTRIBUTED__LOGGING__DISTRIBUTED=warning # ----------------------------------------------------------------------------- # AIQ_SUMMARY_DB= +# ----------------------------------------------------------------------------- +# Artifact byte storage +# Leave AIQ_ARTIFACT_BLOB_PROVIDER unset (or set it to sql) to keep bytes in SQL. +# Set it to s3 for AWS S3, MinIO, Ceph, R2, or another S3-compatible service. +# S3 storage requires the `s3` package extra and a bucket. It does not fall back +# to SQL when an object upload fails. +# ----------------------------------------------------------------------------- +# AIQ_ARTIFACT_BLOB_PROVIDER=sql +# AIQ_ARTIFACT_S3_BUCKET=aiq-artifacts # Required when provider=s3 +# AIQ_ARTIFACT_S3_ENDPOINT_URL= # Unset for AWS; set for MinIO/compatible storage +# AIQ_ARTIFACT_S3_REGION=us-west-2 # Optional +# AIQ_ARTIFACT_S3_PREFIX=artifacts/v1 # Optional; shown value is the default +# Credentials use workload identity or the standard AWS credential chain. +# Local MinIO may set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in deploy/.env. + # ----------------------------------------------------------------------------- # Per-user MCP auth token store (config_web_frag_mcp_auth.yml) # Shared, persistent store the API process (connect) and the Dask worker (job) diff --git a/deploy/Dockerfile b/deploy/Dockerfile index 3c13f928c..737f8725e 100644 --- a/deploy/Dockerfile +++ b/deploy/Dockerfile @@ -76,7 +76,7 @@ COPY deploy/entrypoint.py deploy/start_web.py ./deploy/ # import-fail at runtime. Syncing with the dev group installs those deps from the # frozen lock (still pinned, no re-resolve). TODO(deploy): move the source # packages to a dedicated non-dev group to keep test-only tooling out of the image. -RUN uv sync --frozen +RUN uv sync --frozen --extra s3 # Install workspace packages (without CLI for base) RUN uv pip install --no-deps -e . \ diff --git a/deploy/compose/README.md b/deploy/compose/README.md index 68a96293c..c841042a8 100644 --- a/deploy/compose/README.md +++ b/deploy/compose/README.md @@ -75,6 +75,17 @@ Choose one database configuration in `deploy/.env`: - You can keep the `postgres` service running or remove the `depends_on` block for `aiq-agent` if you want a SQLite-only setup. +### Artifact Storage + +Files generated by sandbox execution, such as charts and CSVs, are captured as +artifacts. Their bytes remain in SQL when `AIQ_ARTIFACT_BLOB_PROVIDER` is unset or +set to `sql`. When the provider is `s3`, artifact bytes are stored in the configured +bucket and SQL stores artifact metadata only. For MinIO, Ceph, R2, or another +compatible service, set `AIQ_ARTIFACT_S3_ENDPOINT_URL`; leave it unset for AWS S3. +Region and prefix are optional, with `artifacts/v1` as the default prefix. +Credentials use the standard AWS credential chain. See `deploy/.env.example` for +the complete variable list. + ## Start Services ### Option 1: Build locally (default) diff --git a/docs/source/architecture/agents/sandbox.md b/docs/source/architecture/agents/sandbox.md index 5b92b8c88..1a09cfb51 100644 --- a/docs/source/architecture/agents/sandbox.md +++ b/docs/source/architecture/agents/sandbox.md @@ -34,6 +34,16 @@ runtime (`.../job/{job_id}/artifacts`), which is also auth-scoped to the job. be written through DeepAgents virtual paths such as `/shared/`; durable binaries (charts, CSVs) are captured by the artifact runtime. +## Artifact Storage + +Metadata for generated files such as charts and CSVs is stored in SQL. File content can +also be stored in SQL, but object storage such as AWS S3 or MinIO is recommended for +production deployments. The selected artifact storage provider determines where the file +content is stored. + +For configuration variables and examples, see [Docker Compose](../../deployment/docker-compose.md#artifact-storage) +and [Production Considerations](../../deployment/production.md#artifact-storage). + ## Operational Notes - High-concurrency Modal runs create one sandbox per job. OpenShell runs share the named diff --git a/docs/source/deployment/docker-compose.md b/docs/source/deployment/docker-compose.md index 535f61ce7..ba140174f 100644 --- a/docs/source/deployment/docker-compose.md +++ b/docs/source/deployment/docker-compose.md @@ -100,6 +100,58 @@ AIQ_CHECKPOINT_DB=/app/data/checkpoints.db When using SQLite, you can optionally remove the `depends_on` block for the `aiq-agent` service since the `postgres` container is no longer needed. +### Artifact Storage + +Artifact metadata always uses the job database. Artifact bytes use SQL BLOB storage +by default. The Compose stack does not provision an AWS S3 bucket; create the bucket +externally, then set: + +```bash +AIQ_ARTIFACT_BLOB_PROVIDER=s3 +AIQ_ARTIFACT_S3_BUCKET=YOUR_BUCKET_NAME +AIQ_ARTIFACT_S3_REGION=us-west-2 +AIQ_ARTIFACT_S3_PREFIX=artifacts/v1 +``` + +The checked-in Compose stack also does not deploy MinIO. To run a local MinIO server +with Docker: + +```bash +export MINIO_CONTAINER=aiq-minio +export MINIO_ROOT_USER=YOUR_ACCESS_KEY +export MINIO_ROOT_PASSWORD=YOUR_SECRET_KEY +export AIQ_ARTIFACT_S3_BUCKET=YOUR_BUCKET_NAME + +docker pull minio/minio +docker run --detach --name "$MINIO_CONTAINER" \ + --publish 9000:9000 --publish 9001:9001 \ + --env MINIO_ROOT_USER --env MINIO_ROOT_PASSWORD \ + --volume aiq-minio-data:/data \ + minio/minio server /data --console-address ":9001" +``` + +Open the MinIO console at `http://localhost:9001` and create the bucket named by +`AIQ_ARTIFACT_S3_BUCKET`. Then add the same bucket and credentials to `deploy/.env`. +With Docker Desktop, the backend container reaches the host through +`host.docker.internal`: + +```bash +AIQ_ARTIFACT_BLOB_PROVIDER=s3 +AIQ_ARTIFACT_S3_BUCKET=YOUR_BUCKET_NAME +AIQ_ARTIFACT_S3_ENDPOINT_URL=http://host.docker.internal:9000 +AWS_ACCESS_KEY_ID=YOUR_ACCESS_KEY +AWS_SECRET_ACCESS_KEY=YOUR_SECRET_KEY +``` + +For another environment, replace the endpoint with an address reachable from the +`aiq-agent` container. + +The bucket is required when the provider is `s3`. Endpoint, region, and prefix are +optional; leave the endpoint unset for AWS S3, and the prefix defaults to +`artifacts/v1`. Configure credentials through workload identity, deployment secrets, or +the standard AWS credential chain. When the provider is `s3`, artifact bytes are stored +in the configured bucket and SQL stores artifact metadata only. + ### Frontend Runtime Settings | Variable | Default | Description | diff --git a/docs/source/deployment/production.md b/docs/source/deployment/production.md index 337de6af5..165599fbb 100644 --- a/docs/source/deployment/production.md +++ b/docs/source/deployment/production.md @@ -49,6 +49,23 @@ pg_dump -U aiq -d aiq_jobs > aiq_jobs_$(date +%Y%m%d).sql pg_dump -U aiq -d aiq_checkpoints > aiq_checkpoints_$(date +%Y%m%d).sql ``` +## Artifact Storage + +Keep artifact metadata in PostgreSQL and use S3-compatible object storage for artifact +bytes in production. SQL BLOB storage remains the default when the provider is unset. + +| Variable | Required | Description | +|----------|----------|-------------| +| `AIQ_ARTIFACT_BLOB_PROVIDER` | No | `sql` by default; set to `s3` for object storage. | +| `AIQ_ARTIFACT_S3_BUCKET` | With S3 | Destination bucket. | +| `AIQ_ARTIFACT_S3_ENDPOINT_URL` | No | Leave unset for AWS S3; set for MinIO, Ceph, R2, or another compatible endpoint. | +| `AIQ_ARTIFACT_S3_REGION` | No | S3 region when required by the provider. | +| `AIQ_ARTIFACT_S3_PREFIX` | No | Object-key prefix; defaults to `artifacts/v1`. | + +Configure credentials through workload identity, deployment secrets, or the standard +AWS credential chain. When the provider is `s3`, artifact bytes are stored in the +configured bucket and SQL stores artifact metadata only. + ## Scaling ### Horizontal Backend Scaling diff --git a/frontends/aiq_api/src/aiq_api/routes/jobs.py b/frontends/aiq_api/src/aiq_api/routes/jobs.py index 96121cb49..2d0ac8641 100644 --- a/frontends/aiq_api/src/aiq_api/routes/jobs.py +++ b/frontends/aiq_api/src/aiq_api/routes/jobs.py @@ -64,6 +64,13 @@ logger = logging.getLogger(__name__) +def _validate_artifact_store(db_url: str) -> None: + """Validate configured artifact storage during API startup.""" + from aiq_agent.agents.deep_researcher.sandbox.artifacts import build_artifact_store + + build_artifact_store(db_url).validate() + + def _int_env(name: str, default: int) -> int: """Read a non-negative integer ops knob from the environment. @@ -569,7 +576,9 @@ async def list_data_sources() -> list[DataSource]: db_url[:50], default_expiry_seconds, ) - await asyncio.get_running_loop().run_in_executor(None, ensure_job_access_table, db_url) + loop = asyncio.get_running_loop() + await loop.run_in_executor(None, ensure_job_access_table, db_url) + await loop.run_in_executor(None, _validate_artifact_store, db_url) @app.get("/health", tags=["health"], summary="Health check") async def health_check(): @@ -898,12 +907,12 @@ async def get_job_state(job_id: str) -> JobStateResponse: ) async def list_job_artifacts(job_id: str) -> dict: """List durable artifact metadata for a job (no bytes).""" - from aiq_agent.agents.deep_researcher.sandbox.artifacts import SqlArtifactStore + from aiq_agent.agents.deep_researcher.sandbox.artifacts import build_artifact_store principal = require_verified_principal() await authorize_job_access(job_store, db_url, job_id, principal) - store = SqlArtifactStore(db_url) + store = build_artifact_store(db_url) artifacts = await asyncio.to_thread(store.list, job_id) # Exclude storage internals (storage_uri embeds the db_url, which may carry # credentials/hostnames; sandbox_path is an internal layout detail) from the @@ -922,12 +931,12 @@ async def list_job_artifacts(job_id: str) -> dict: ) async def get_job_artifact_content(job_id: str, artifact_id: str) -> StreamingResponse: """Stream an artifact's bytes (auth-scoped to the owning job).""" - from aiq_agent.agents.deep_researcher.sandbox.artifacts import SqlArtifactStore + from aiq_agent.agents.deep_researcher.sandbox.artifacts import build_artifact_store principal = require_verified_principal() await authorize_job_access(job_store, db_url, job_id, principal) - store = SqlArtifactStore(db_url) + store = build_artifact_store(db_url) artifact = await asyncio.to_thread(store.get, job_id, artifact_id) if artifact is None: raise HTTPException(404, f"Artifact not found: {artifact_id}") @@ -1268,10 +1277,10 @@ def _do_cleanup() -> tuple[int, int, int]: # Artifact retention shares the job expiry boundary (best-effort; the artifacts # table only exists when artifact capture has been used). try: - from aiq_agent.agents.deep_researcher.sandbox.artifacts import SqlArtifactStore + from aiq_agent.agents.deep_researcher.sandbox.artifacts import build_artifact_store artifacts_deleted = await loop.run_in_executor( - None, lambda: SqlArtifactStore(db_url).cleanup_old_artifacts(retention_seconds) + None, lambda: build_artifact_store(db_url).cleanup_old_artifacts(retention_seconds) ) if artifacts_deleted: logger.info("Artifact cleanup: %d old artifacts removed", artifacts_deleted) diff --git a/frontends/aiq_api/tests/test_job_submit_data_sources.py b/frontends/aiq_api/tests/test_job_submit_data_sources.py index 1d0045ee4..42f683e15 100644 --- a/frontends/aiq_api/tests/test_job_submit_data_sources.py +++ b/frontends/aiq_api/tests/test_job_submit_data_sources.py @@ -65,6 +65,7 @@ async def submit_app(monkeypatch): submitted_job = AsyncMock(return_value="job-1") monkeypatch.setattr(jobs_routes, "_start_periodic_cleanup", MagicMock()) + monkeypatch.setattr(jobs_routes, "_validate_artifact_store", MagicMock()) agent_config = AgentConfig( class_path="aiq_agent.agents.deep_researcher.agent.DeepResearcherAgent", @@ -127,6 +128,27 @@ async def _filtered_get_tools(*, tool_names, wrapper_type): # noqa: ARG001 - mi return app, submitted_job, builder +@pytest.mark.asyncio +async def test_route_registration_validates_artifact_store(submit_app): + import aiq_api.routes.jobs as jobs_routes + + _app, _submitted_job, _builder = submit_app + + jobs_routes._validate_artifact_store.assert_called_once_with("sqlite:///./test.db") + + +def test_artifact_store_validation_propagates_failure(monkeypatch): + import aiq_api.routes.jobs as jobs_routes + from aiq_agent.agents.deep_researcher.sandbox import artifacts + + store = MagicMock() + store.validate.side_effect = RuntimeError("storage unavailable") + monkeypatch.setattr(artifacts, "build_artifact_store", MagicMock(return_value=store)) + + with pytest.raises(RuntimeError, match="storage unavailable"): + jobs_routes._validate_artifact_store("sqlite:///./test.db") + + @pytest.mark.asyncio async def test_submit_job_forwards_selected_data_sources(submit_app): app, submitted_job, builder = submit_app diff --git a/pyproject.toml b/pyproject.toml index fe0175630..950861ba0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,6 +54,9 @@ dependencies = [ ] [project.optional-dependencies] +s3 = [ + "boto3>=1.35.0,<2", +] dev = [ "pytest>=7.4.0", "pytest-asyncio>=0.21.0", diff --git a/src/aiq_agent/agents/deep_researcher/deepagents_runtime.py b/src/aiq_agent/agents/deep_researcher/deepagents_runtime.py index 33be229e8..9081391b0 100644 --- a/src/aiq_agent/agents/deep_researcher/deepagents_runtime.py +++ b/src/aiq_agent/agents/deep_researcher/deepagents_runtime.py @@ -273,12 +273,12 @@ def _maybe_build_artifact_manager( if capture is None or not getattr(capture, "enabled", False): return None from .sandbox.artifacts import ArtifactManager - from .sandbox.artifacts import SqlArtifactStore + from .sandbox.artifacts import build_artifact_store return ArtifactManager( job_id=job_id, backend=provider, - store=SqlArtifactStore(artifact_db_url), + store=build_artifact_store(artifact_db_url), config=capture, artifact_dir=artifact_dir, emit=artifact_emit, diff --git a/src/aiq_agent/agents/deep_researcher/sandbox/README.md b/src/aiq_agent/agents/deep_researcher/sandbox/README.md index 3c0ed2b1d..baa502b4a 100644 --- a/src/aiq_agent/agents/deep_researcher/sandbox/README.md +++ b/src/aiq_agent/agents/deep_researcher/sandbox/README.md @@ -63,7 +63,9 @@ plus `/shared/` for durable text. Binary artifacts are harvested host-side via | `providers/openshell.py` | OpenShell provider (enterprise/on-prem). Lazy, ad-hoc deps; policy requires a named sandbox. | | `artifacts/models.py` | `Artifact` record (id, mime, sha256, size, provenance, status). Metadata only. | | `artifacts/manifest.py` | `manifest.json` schema + parser. | -| `artifacts/store.py` | `SqlArtifactStore` on the shared job `db_url` (metadata table + capped BLOB; pluggable to S3). | +| `artifacts/store.py` | `SqlArtifactStore` coordinates SQL metadata with the configured byte provider. | +| `artifacts/blob_store.py` | Byte adapters for SQL BLOBs and S3-compatible object storage. | +| `artifacts/factory.py` | Builds the application store from `AIQ_ARTIFACT_*` environment variables. | | `artifacts/manager.py` | Harvest pipeline: manifest-first + scan, path-traversal confinement, MIME-from-bytes, SVG sanitize, render-gate, quotas, dedup, store-then-emit, `artifact://` resolution. | ## Adding a provider (the whole surface) @@ -154,7 +156,7 @@ is lifted into `providers.modal`. - Once at the end of the agent run (`agent.run()` -> `ArtifactManager.final_harvest`), the `ArtifactManager` pulls bytes via `download_files`, runs the validation pipeline (path-traversal confinement -> extension allowlist -> size cap -> MIME-from-bytes/spoof - reject -> quota -> SVG sanitize -> sha256), stores via `SqlArtifactStore`, then emits an + reject -> quota -> SVG sanitize -> sha256), stores via `ArtifactStore`, then emits an `artifact` SSE event (`to_sse_payload`, metadata + `content_url`, never bytes). - Failed or cancelled runs are not harvested in the current implementation. - Reports reference artifacts as `![caption](artifact://)`; the report @@ -248,13 +250,52 @@ to delegate uploads to the official adapter and validate the upstream argv fix always use AI-Q's bounded shim so realpath confinement and pre-transfer size checks remain in force. Once the upstream adapter provides equivalent guards, drop the shim and toggle. +## Artifact byte storage + +SQL BLOB storage remains the default when `AIQ_ARTIFACT_BLOB_PROVIDER` is unset or +set to `sql`. Production deployments can set it to `s3` to store bytes in AWS S3 or +an S3-compatible service while retaining artifact metadata in the job database. + +AWS S3: + +```bash +AIQ_ARTIFACT_BLOB_PROVIDER=s3 +AIQ_ARTIFACT_S3_BUCKET=aiq-artifacts +AIQ_ARTIFACT_S3_REGION=us-west-2 +AIQ_ARTIFACT_S3_PREFIX=artifacts/v1 +``` + +MinIO or another S3-compatible service uses the same provider and additionally sets +the custom endpoint: + +```bash +AIQ_ARTIFACT_BLOB_PROVIDER=s3 +AIQ_ARTIFACT_S3_BUCKET=aiq-artifacts +AIQ_ARTIFACT_S3_ENDPOINT_URL=http://minio:9000 +AIQ_ARTIFACT_S3_REGION=us-east-1 +AIQ_ARTIFACT_S3_PREFIX=artifacts/v1 +``` + +`AIQ_ARTIFACT_S3_BUCKET` is required for S3 storage. The endpoint is optional: leave +it unset for AWS S3 and set it for MinIO, Ceph, R2, or another compatible endpoint. +The region and prefix are optional; the prefix defaults to `artifacts/v1`. Credentials +come from workload identity, deployment secrets, or the standard AWS credential chain. +For local MinIO, `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` can hold the local +MinIO credentials. Install the optional dependency with `uv sync --extra s3`. + +Selecting `s3` does not automatically fall back to SQL if object storage fails. The selected +provider applies to the whole application. Artifact cleanup follows the retention period and +removes object bytes before SQL metadata. + +Custom endpoints use path-style bucket addressing for MinIO compatibility. + ## Operational knobs - `AIQ_MAX_SANDBOXES_PER_PRINCIPAL` / `AIQ_MAX_SANDBOXES_GLOBAL` (default-off): submit-path concurrency/cost caps for sandbox-enabled jobs. - `AIQ_OPENSHELL_ADAPTER_FILE_TRANSFER` (default-off): route OpenShell uploads through the official adapter instead of the env-free shim (see OpenShell gotcha above). -- Artifact retention reuses the job-expiry periodic cleanup (`expiry_seconds`). +- Artifact retention reuses the existing periodic cleanup (`expiry_seconds`). - In-container OpenShell log verbosity (opt-in): `agent.execute()` calls and their output are already logged on the AI-Q side (the `execute` tool-call events). To also see what runs inside the OpenShell container, rebuild the sandbox image with a higher `RUST_LOG`: diff --git a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/__init__.py b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/__init__.py index 9bf1578ad..b5e50a773 100644 --- a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/__init__.py +++ b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/__init__.py @@ -17,6 +17,10 @@ from __future__ import annotations +from .blob_store import ArtifactBlobStore +from .blob_store import S3ArtifactBlobStore +from .blob_store import SqlArtifactBlobStore +from .factory import build_artifact_store from .manager import ArtifactManager from .manifest import Manifest from .manifest import ManifestEntry @@ -38,6 +42,10 @@ "ManifestEntry", "parse_manifest", "ArtifactStore", + "ArtifactBlobStore", + "SqlArtifactBlobStore", + "S3ArtifactBlobStore", + "build_artifact_store", "LocalArtifactStore", "SqlArtifactStore", "ArtifactManager", diff --git a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/blob_store.py b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/blob_store.py new file mode 100644 index 000000000..6fb8632dc --- /dev/null +++ b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/blob_store.py @@ -0,0 +1,164 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Byte storage adapters for durable sandbox artifacts.""" + +from __future__ import annotations + +from abc import ABC +from abc import abstractmethod +from collections.abc import Iterator +from typing import Any +from urllib.parse import urlparse + +from .models import Artifact + +_READ_CHUNK_BYTES = 1 << 20 + + +class ArtifactBlobStore(ABC): + """Provider interface for artifact bytes; metadata remains in SQL.""" + + @property + @abstractmethod + def scheme(self) -> str: + """URI scheme handled by this provider.""" + + @abstractmethod + def make_uri(self, artifact: Artifact) -> str: + """Return the durable URI to persist in artifact metadata.""" + + @abstractmethod + def put(self, artifact: Artifact, data: bytes) -> bytes | None: + """Write bytes externally, or return bytes stored with SQL metadata.""" + + @abstractmethod + def open_bytes(self, artifact: Artifact) -> Iterator[bytes]: + """Stream artifact bytes from the provider.""" + + @abstractmethod + def delete(self, artifact: Artifact) -> None: + """Delete artifact bytes. Missing bytes must be treated as success.""" + + @abstractmethod + def validate(self) -> None: + """Raise when the configured byte storage is unavailable.""" + + +class SqlArtifactBlobStore(ArtifactBlobStore): + """Store artifact bytes in the existing SQL ``content`` column.""" + + def __init__(self, engine: Any) -> None: + self._engine = engine + + @property + def scheme(self) -> str: + return "db" + + def make_uri(self, artifact: Artifact) -> str: + return f"db://artifacts/{artifact.artifact_id}" + + def put(self, artifact: Artifact, data: bytes) -> bytes: + return data + + def open_bytes(self, artifact: Artifact) -> Iterator[bytes]: + from sqlalchemy import text + + with self._engine.connect() as conn: + row = conn.execute( + text("SELECT content FROM artifacts WHERE job_id = :job_id AND artifact_id = :artifact_id"), + {"job_id": artifact.job_id, "artifact_id": artifact.artifact_id}, + ).fetchone() + if row is None or row[0] is None: + return + data: bytes = row[0] + for start in range(0, len(data), _READ_CHUNK_BYTES): + yield data[start : start + _READ_CHUNK_BYTES] + + def delete(self, artifact: Artifact) -> None: + # Inline bytes are deleted with their SQL metadata row. + pass + + def validate(self) -> None: + pass + + +class S3ArtifactBlobStore(ArtifactBlobStore): + """Store bytes in AWS S3 or an S3-compatible service such as MinIO.""" + + def __init__( + self, + *, + bucket: str, + prefix: str = "artifacts/v1", + endpoint_url: str | None = None, + region: str | None = None, + client: Any | None = None, + ) -> None: + if not bucket: + raise ValueError("AIQ_ARTIFACT_S3_BUCKET is required when the artifact blob provider is s3") + self.bucket = bucket + self.prefix = prefix.strip("/") + if client is None: + try: + import boto3 + from botocore.config import Config + except ImportError as exc: + raise RuntimeError( + "S3 artifact storage requires the optional 's3' dependency: install aiq-agent[s3]" + ) from exc + client = boto3.client( + "s3", + endpoint_url=endpoint_url or None, + region_name=region or None, + config=Config( + connect_timeout=10, + read_timeout=30, + retries={"max_attempts": 3, "mode": "standard"}, + # MinIO and other custom endpoints do not necessarily provide + # bucket-name DNS. Path style works consistently for them. + s3={"addressing_style": "path"} if endpoint_url else {}, + ), + ) + self._client = client + + @property + def scheme(self) -> str: + return "s3" + + def make_uri(self, artifact: Artifact) -> str: + key = "/".join(part for part in (self.prefix, artifact.job_id, artifact.artifact_id) if part) + return f"s3://{self.bucket}/{key}" + + def put(self, artifact: Artifact, data: bytes) -> None: + bucket, key = self._location(artifact.storage_uri) + self._client.put_object( + Bucket=bucket, + Key=key, + Body=data, + ContentType=artifact.mime_type, + Metadata={"sha256": artifact.sha256}, + ) + + def open_bytes(self, artifact: Artifact) -> Iterator[bytes]: + bucket, key = self._location(artifact.storage_uri) + response = self._client.get_object(Bucket=bucket, Key=key) + body = response["Body"] + try: + yield from body.iter_chunks(chunk_size=_READ_CHUNK_BYTES) + finally: + body.close() + + def delete(self, artifact: Artifact) -> None: + bucket, key = self._location(artifact.storage_uri) + self._client.delete_object(Bucket=bucket, Key=key) + + def validate(self) -> None: + self._client.head_bucket(Bucket=self.bucket) + + @staticmethod + def _location(storage_uri: str) -> tuple[str, str]: + parsed = urlparse(storage_uri) + if parsed.scheme != "s3" or not parsed.netloc or not parsed.path.lstrip("/"): + raise ValueError(f"Invalid S3 artifact URI: {storage_uri}") + return parsed.netloc, parsed.path.lstrip("/") diff --git a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/factory.py b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/factory.py new file mode 100644 index 000000000..f91d94832 --- /dev/null +++ b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/factory.py @@ -0,0 +1,49 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Configured construction for the application-level artifact store.""" + +from __future__ import annotations + +import os +from functools import lru_cache + +from .blob_store import S3ArtifactBlobStore +from .store import ArtifactStore +from .store import SqlArtifactStore + + +def build_artifact_store(db_url: str) -> ArtifactStore: + """Build the SQL metadata store with the configured byte provider.""" + provider = os.environ.get("AIQ_ARTIFACT_BLOB_PROVIDER", "sql").strip().lower() + return _build_artifact_store( + db_url, + provider, + os.environ.get("AIQ_ARTIFACT_S3_BUCKET", ""), + os.environ.get("AIQ_ARTIFACT_S3_ENDPOINT_URL"), + os.environ.get("AIQ_ARTIFACT_S3_REGION"), + os.environ.get("AIQ_ARTIFACT_S3_PREFIX", "artifacts/v1"), + ) + + +@lru_cache(maxsize=16) +def _build_artifact_store( + db_url: str, + provider: str, + bucket: str, + endpoint_url: str | None, + region: str | None, + prefix: str, +) -> ArtifactStore: + """Construct and cache one configured store per process and configuration.""" + if provider == "sql": + return SqlArtifactStore(db_url) + if provider == "s3": + blob_store = S3ArtifactBlobStore( + bucket=bucket, + endpoint_url=endpoint_url, + region=region, + prefix=prefix, + ) + return SqlArtifactStore(db_url, blob_store=blob_store) + raise ValueError(f"Unsupported AIQ_ARTIFACT_BLOB_PROVIDER: {provider!r}; expected 'sql' or 's3'") diff --git a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/store.py b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/store.py index ba89c566d..286763447 100644 --- a/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/store.py +++ b/src/aiq_agent/agents/deep_researcher/sandbox/artifacts/store.py @@ -15,15 +15,16 @@ """Durable artifact storage. -Metadata and bytes are stored together in an ``artifacts`` table on the shared job -``db_url`` (the same database used by the job/event stores), keyed by -``artifact_id``. This is the byte-locality fix: the Dask worker (writer) and the API -process (reader) both reach artifacts with no shared filesystem. +Artifact metadata is stored in an ``artifacts`` table on the shared job ``db_url`` +(the same database used by the job/event stores), keyed by ``artifact_id``. Artifact +bytes are stored either in the table's ``content`` BLOB column or in the configured +S3-compatible object store. The Dask worker (writer) and API process (reader) use the +same configured storage provider, so no shared filesystem is required. - Metadata columns are queryable and listed in the UI/CLI. -- Bytes live in a size-capped ``content`` BLOB column for milestone 1. -- ``storage_uri`` stays abstract so a production deployment can move bytes to - object storage (S3) without changing callers. +- SQL BLOB storage remains the default for local development. +- Production deployments can store bytes in S3-compatible object storage without + changing callers. """ from __future__ import annotations @@ -36,14 +37,14 @@ from collections.abc import Iterator from typing import Any +from .blob_store import ArtifactBlobStore +from .blob_store import SqlArtifactBlobStore from .models import Artifact from .models import ArtifactProvenance from .models import ArtifactStatus logger = logging.getLogger(__name__) -_READ_CHUNK_BYTES = 1 << 20 # 1 MiB streaming chunk - def _normalize_db_url(db_url: str) -> str: """Normalize a SQLAlchemy URL to a sync driver (mirrors the event store). @@ -90,11 +91,15 @@ def delete_job(self, job_id: str) -> int: def cleanup_old_artifacts(self, retention_seconds: int) -> int: """Delete artifacts older than the retention period. Returns count removed.""" + @abstractmethod + def validate(self) -> None: + """Raise when the configured artifact storage is unavailable.""" + class SqlArtifactStore(ArtifactStore): - """SQL-backed store (SQLite/Postgres) on the shared job ``db_url``. + """SQL-backed metadata store (SQLite/Postgres) on the shared job ``db_url``. - Bytes live in a capped BLOB column. Dedup is per-job by content digest so + Bytes use the configured blob store. Dedup is per-job by content digest so harvest is idempotent across job retries. """ @@ -102,7 +107,12 @@ class SqlArtifactStore(ArtifactStore): _initialized: set[str] = set() _lock = threading.Lock() - def __init__(self, db_url: str = "sqlite+aiosqlite:///./jobs.db") -> None: + def __init__( + self, + db_url: str = "sqlite+aiosqlite:///./jobs.db", + *, + blob_store: ArtifactBlobStore | None = None, + ) -> None: """Bind to the shared job database and ensure the artifacts table exists. Args: @@ -111,6 +121,7 @@ def __init__(self, db_url: str = "sqlite+aiosqlite:///./jobs.db") -> None: self.db_url = db_url self._engine = self._get_engine(db_url) self._ensure_table() + self._blob_store = blob_store or SqlArtifactBlobStore(self._engine) @classmethod def _get_engine(cls, db_url: str) -> Any: @@ -198,15 +209,30 @@ def put(self, artifact: Artifact, data: bytes) -> Artifact: logger.debug("Artifact dedup hit for job=%s sha=%s", artifact.job_id, artifact.sha256[:12]) return existing - from sqlalchemy import text - - stored = artifact.model_copy( + prepared = artifact.model_copy( update={ # Logical location only — never embed db_url (it may carry credentials). - "storage_uri": f"db://artifacts/{artifact.artifact_id}", - "status": ArtifactStatus.AVAILABLE, + "storage_uri": self._blob_store.make_uri(artifact), } ) + try: + content = self._blob_store.put(prepared, data) + stored = prepared.model_copy(update={"status": ArtifactStatus.AVAILABLE}) + self._insert_metadata(stored, content=content) + except Exception: + logger.exception("Artifact storage failed for artifact=%s", prepared.artifact_id) + try: + self._blob_store.delete(prepared) + except Exception: + logger.warning("Failed to remove artifact bytes after storage failure: %s", prepared.storage_uri) + raise + + return stored + + def _insert_metadata(self, artifact: Artifact, *, content: bytes | None) -> None: + """Insert one artifact metadata row and optional SQL-backed content.""" + from sqlalchemy import text + with self._engine.connect() as conn: conn.execute( text( @@ -217,42 +243,33 @@ def put(self, artifact: Artifact, data: bytes) -> Artifact: ":source_tool_call_id, :provenance, :status, :content)" ), { - "artifact_id": stored.artifact_id, - "job_id": stored.job_id, - "kind": stored.kind.value, - "mime_type": stored.mime_type, - "filename": stored.filename, - "sandbox_path": stored.sandbox_path, - "storage_uri": stored.storage_uri, - "sha256": stored.sha256, - "size_bytes": stored.size_bytes, - "title": stored.title, - "caption": stored.caption, - "inline": stored.inline, - "workflow": stored.workflow, - "source_tool_call_id": stored.source_tool_call_id, - "provenance": stored.provenance.model_dump_json(), - "status": stored.status.value, - "content": data, + "artifact_id": artifact.artifact_id, + "job_id": artifact.job_id, + "kind": artifact.kind.value, + "mime_type": artifact.mime_type, + "filename": artifact.filename, + "sandbox_path": artifact.sandbox_path, + "storage_uri": artifact.storage_uri, + "sha256": artifact.sha256, + "size_bytes": artifact.size_bytes, + "title": artifact.title, + "caption": artifact.caption, + "inline": artifact.inline, + "workflow": artifact.workflow, + "source_tool_call_id": artifact.source_tool_call_id, + "provenance": artifact.provenance.model_dump_json(), + "status": artifact.status.value, + "content": content, }, ) conn.commit() - return stored def open_bytes(self, job_id: str, artifact_id: str) -> Iterator[bytes]: - """Yield an artifact's bytes in 1 MiB chunks, or nothing if not found.""" - from sqlalchemy import text - - with self._engine.connect() as conn: - row = conn.execute( - text("SELECT content FROM artifacts WHERE job_id = :job_id AND artifact_id = :artifact_id"), - {"job_id": job_id, "artifact_id": artifact_id}, - ).fetchone() - if row is None or row[0] is None: + """Yield an artifact's bytes from the configured provider, or nothing if not found.""" + artifact = self.get(job_id, artifact_id) + if artifact is None: return - data: bytes = row[0] - for start in range(0, len(data), _READ_CHUNK_BYTES): - yield data[start : start + _READ_CHUNK_BYTES] + yield from self._blob_store.open_bytes(artifact) def get(self, job_id: str, artifact_id: str) -> Artifact | None: """Return artifact metadata for the job, or ``None`` if not found.""" @@ -302,13 +319,8 @@ def list(self, job_id: str) -> list[Artifact]: return [_row_to_artifact(row) for row in rows] def delete_job(self, job_id: str) -> int: - """Delete all artifacts for the job and return the number removed.""" - from sqlalchemy import text - - with self._engine.connect() as conn: - result = conn.execute(text("DELETE FROM artifacts WHERE job_id = :job_id"), {"job_id": job_id}) - conn.commit() - return result.rowcount + """Delete a job's bytes first and remove metadata after each success.""" + return sum(self._delete_artifact(artifact) for artifact in self.list(job_id)) def cleanup_old_artifacts(self, retention_seconds: int) -> int: """Delete artifacts older than the retention window and return the count. @@ -324,17 +336,54 @@ def cleanup_old_artifacts(self, retention_seconds: int) -> int: is_postgres = _normalize_db_url(self.db_url).startswith("postgresql") with self._engine.connect() as conn: if is_postgres: - result = conn.execute( - text("DELETE FROM artifacts WHERE created_at < NOW() - :seconds * INTERVAL '1 second'"), - {"seconds": retention_seconds}, + rows = ( + conn.execute( + text( + f"SELECT {_META_COLUMNS} FROM artifacts " + "WHERE created_at < NOW() - :seconds * INTERVAL '1 second'" + ), + {"seconds": retention_seconds}, + ) + .mappings() + .fetchall() ) else: + rows = ( + conn.execute( + text(f"SELECT {_META_COLUMNS} FROM artifacts WHERE created_at < datetime('now', :interval)"), + {"interval": f"-{retention_seconds} seconds"}, + ) + .mappings() + .fetchall() + ) + return sum(self._delete_artifact(_row_to_artifact(row)) for row in rows) + + def validate(self) -> None: + """Validate the configured byte storage.""" + self._blob_store.validate() + + def _delete_artifact(self, artifact: Artifact) -> int: + """Delete one artifact's bytes and then its metadata row.""" + from sqlalchemy import text + + try: + self._blob_store.delete(artifact) + except Exception: + logger.exception("Artifact byte deletion failed; retaining metadata for %s", artifact.artifact_id) + return 0 + try: + with self._engine.connect() as conn: result = conn.execute( - text("DELETE FROM artifacts WHERE created_at < datetime('now', :interval)"), - {"interval": f"-{retention_seconds} seconds"}, + text("DELETE FROM artifacts WHERE job_id = :job_id AND artifact_id = :artifact_id"), + {"job_id": artifact.job_id, "artifact_id": artifact.artifact_id}, ) - conn.commit() - return result.rowcount + conn.commit() + return result.rowcount + except Exception: + logger.exception( + "Artifact metadata deletion failed; retaining metadata for retry: %s", artifact.artifact_id + ) + return 0 # Backwards-friendly alias; local development and single-node use the same SQL store. diff --git a/tests/aiq_agent/agents/deep_researcher/sandbox/test_artifacts.py b/tests/aiq_agent/agents/deep_researcher/sandbox/test_artifacts.py index 6a5628ed2..eadccea91 100644 --- a/tests/aiq_agent/agents/deep_researcher/sandbox/test_artifacts.py +++ b/tests/aiq_agent/agents/deep_researcher/sandbox/test_artifacts.py @@ -14,18 +14,25 @@ # limitations under the License. """Tests for the artifact runtime: manifest parsing, MIME sniffing, the harvest -validation pipeline (traversal/extension/size/quota/dedup/scan), and the SQL store.""" +validation pipeline (traversal/extension/size/quota/dedup/scan), and SQL or S3-compatible stores.""" from __future__ import annotations import json +from hashlib import sha256 +from io import BytesIO from types import SimpleNamespace from typing import Any +import pytest + from aiq_agent.agents.deep_researcher.sandbox.artifacts import Artifact from aiq_agent.agents.deep_researcher.sandbox.artifacts import ArtifactKind from aiq_agent.agents.deep_researcher.sandbox.artifacts import ArtifactManager +from aiq_agent.agents.deep_researcher.sandbox.artifacts import ArtifactStatus +from aiq_agent.agents.deep_researcher.sandbox.artifacts import S3ArtifactBlobStore from aiq_agent.agents.deep_researcher.sandbox.artifacts import SqlArtifactStore +from aiq_agent.agents.deep_researcher.sandbox.artifacts import build_artifact_store from aiq_agent.agents.deep_researcher.sandbox.artifacts import parse_manifest from aiq_agent.agents.deep_researcher.sandbox.artifacts.manager import _normalize_posix from aiq_agent.agents.deep_researcher.sandbox.artifacts.manager import _sniff_mime @@ -33,6 +40,7 @@ _ARTIFACT_DIR = "/workspace/aiq-artifacts" _PNG = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64 +_PNG_SHA256 = sha256(_PNG).hexdigest() class _FakeBackend: @@ -55,7 +63,7 @@ def _manifest_bytes(path: str, kind: str = "image") -> bytes: return json.dumps({"version": 1, "artifacts": [{"path": path, "kind": kind, "inline": True}]}).encode("utf-8") -def _make_manager(store: SqlArtifactStore, files: dict[str, bytes], **capture: Any) -> tuple[ArtifactManager, list]: +def _make_manager(store: Any, files: dict[str, bytes], **capture: Any) -> tuple[ArtifactManager, list]: emitted: list = [] config = ArtifactCaptureConfig(enabled=True, **capture) manager = ArtifactManager( @@ -220,7 +228,7 @@ def _artifact(self) -> Artifact: filename="chart.png", sandbox_path=f"{_ARTIFACT_DIR}/chart.png", storage_uri="", - sha256="d" * 64, + sha256=_PNG_SHA256, size_bytes=len(_PNG), ) @@ -243,6 +251,200 @@ def test_dedup_by_digest(self, tmp_path: Any) -> None: assert again.artifact_id == first.artifact_id assert len(store.list("job-1")) == 1 + def test_cleanup_removes_old_artifacts(self, tmp_path: Any) -> None: + from sqlalchemy import text + + store = SqlArtifactStore(f"sqlite:///{tmp_path}/jobs.db") + stored = store.put(self._artifact(), _PNG) + with store._engine.connect() as conn: + conn.execute( + text("UPDATE artifacts SET created_at = datetime('now', '-2 seconds') WHERE artifact_id = :id"), + {"id": stored.artifact_id}, + ) + conn.commit() + + assert store.cleanup_old_artifacts(1) == 1 + assert store.get(stored.job_id, stored.artifact_id) is None + + +class _FakeStreamingBody: + def __init__(self, data: bytes) -> None: + self._stream = BytesIO(data) + self.closed = False + + def iter_chunks(self, chunk_size: int) -> Any: + while chunk := self._stream.read(chunk_size): + yield chunk + + def close(self) -> None: + self.closed = True + self._stream.close() + + +class _FakeS3Client: + def __init__(self) -> None: + self.objects: dict[tuple[str, str], bytes] = {} + self.fail_put = False + self.fail_delete = False + self.head_bucket_calls: list[str] = [] + + def put_object(self, **kwargs: Any) -> None: + if self.fail_put: + raise RuntimeError("upload failed") + location = (kwargs["Bucket"], kwargs["Key"]) + self.objects[location] = kwargs["Body"] + + def get_object(self, **kwargs: Any) -> dict[str, Any]: + return {"Body": _FakeStreamingBody(self.objects[(kwargs["Bucket"], kwargs["Key"])])} + + def delete_object(self, **kwargs: Any) -> None: + if self.fail_delete: + raise RuntimeError("delete failed") + location = (kwargs["Bucket"], kwargs["Key"]) + self.objects.pop(location, None) + + def head_bucket(self, **kwargs: Any) -> None: + self.head_bucket_calls.append(kwargs["Bucket"]) + + +class TestS3Store: + def _artifact(self) -> Artifact: + return Artifact( + artifact_id="art_" + "a" * 32, + job_id="job-1", + kind=ArtifactKind.IMAGE, + mime_type="image/png", + filename="chart.png", + sandbox_path=f"{_ARTIFACT_DIR}/chart.png", + storage_uri="", + sha256=_PNG_SHA256, + size_bytes=len(_PNG), + ) + + def _store( + self, + tmp_path: Any, + client: _FakeS3Client, + ) -> SqlArtifactStore: + blob_store = S3ArtifactBlobStore( + bucket="aiq-artifacts", + prefix="artifacts/v1", + endpoint_url="http://minio:9000", + client=client, + ) + return SqlArtifactStore(f"sqlite:///{tmp_path}/jobs.db", blob_store=blob_store) + + def test_put_open_and_delete(self, tmp_path: Any) -> None: + client = _FakeS3Client() + store = self._store(tmp_path, client) + + stored = store.put(self._artifact(), _PNG) + + assert stored.storage_uri == f"s3://aiq-artifacts/artifacts/v1/job-1/{stored.artifact_id}" + assert stored.status == ArtifactStatus.AVAILABLE + assert b"".join(store.open_bytes("job-1", stored.artifact_id)) == _PNG + assert store.delete_job("job-1") == 1 + assert client.objects == {} + assert store.get("job-1", stored.artifact_id) is None + + def test_validate_checks_configured_bucket(self, tmp_path: Any) -> None: + client = _FakeS3Client() + store = self._store(tmp_path, client) + + store.validate() + + assert client.head_bucket_calls == ["aiq-artifacts"] + + def test_s3_bytes_leave_sql_content_null(self, tmp_path: Any) -> None: + from sqlalchemy import text + + store = self._store(tmp_path, _FakeS3Client()) + stored = store.put(self._artifact(), _PNG) + with store._engine.connect() as conn: + content = conn.execute( + text("SELECT content FROM artifacts WHERE artifact_id = :artifact_id"), + {"artifact_id": stored.artifact_id}, + ).scalar_one() + assert content is None + + def test_failed_upload_removes_metadata(self, tmp_path: Any) -> None: + client = _FakeS3Client() + client.fail_put = True + store = self._store(tmp_path, client) + artifact = self._artifact() + + with pytest.raises(RuntimeError, match="upload failed"): + store.put(artifact, _PNG) + + assert store.get(artifact.job_id, artifact.artifact_id) is None + assert b"".join(store.open_bytes(artifact.job_id, artifact.artifact_id)) == b"" + + def test_failed_delete_retains_metadata_for_retry(self, tmp_path: Any) -> None: + client = _FakeS3Client() + store = self._store(tmp_path, client) + stored = store.put(self._artifact(), _PNG) + client.fail_delete = True + + assert store.delete_job(stored.job_id) == 0 + assert store.get(stored.job_id, stored.artifact_id) is not None + + def test_failed_metadata_delete_is_retried(self, tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: + client = _FakeS3Client() + store = self._store(tmp_path, client) + stored = store.put(self._artifact(), _PNG) + + def fail_connect() -> None: + raise RuntimeError("database unavailable") + + with monkeypatch.context() as patch: + patch.setattr(store._engine, "connect", fail_connect) + assert store._delete_artifact(stored) == 0 + + assert client.objects == {} + assert store.get(stored.job_id, stored.artifact_id) is not None + assert store._delete_artifact(stored) == 1 + assert store.get(stored.job_id, stored.artifact_id) is None + + +class TestArtifactStoreFactory: + def test_sql_is_the_default_provider(self, tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("AIQ_ARTIFACT_BLOB_PROVIDER", raising=False) + + store = build_artifact_store(f"sqlite:///{tmp_path}/jobs.db") + + assert isinstance(store, SqlArtifactStore) + assert store._blob_store.scheme == "db" + + def test_s3_provider_uses_configured_blob_store( + self, + tmp_path: Any, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + from aiq_agent.agents.deep_researcher.sandbox.artifacts import factory + + fake_blob_store = S3ArtifactBlobStore(bucket="aiq-artifacts", client=_FakeS3Client()) + monkeypatch.setenv("AIQ_ARTIFACT_BLOB_PROVIDER", "s3") + monkeypatch.setenv("AIQ_ARTIFACT_S3_BUCKET", "aiq-artifacts") + monkeypatch.setattr(factory, "S3ArtifactBlobStore", lambda **_kwargs: fake_blob_store) + factory._build_artifact_store.cache_clear() + + store = build_artifact_store(f"sqlite:///{tmp_path}/jobs.db") + + assert isinstance(store, SqlArtifactStore) + assert store._blob_store is fake_blob_store + factory._build_artifact_store.cache_clear() + + def test_rejects_unknown_provider(self, tmp_path: Any, monkeypatch: pytest.MonkeyPatch) -> None: + from aiq_agent.agents.deep_researcher.sandbox.artifacts import factory + + monkeypatch.setenv("AIQ_ARTIFACT_BLOB_PROVIDER", "unknown") + factory._build_artifact_store.cache_clear() + + with pytest.raises(ValueError, match="Unsupported AIQ_ARTIFACT_BLOB_PROVIDER"): + build_artifact_store(f"sqlite:///{tmp_path}/jobs.db") + + factory._build_artifact_store.cache_clear() + class TestEnsureInlineArtifactsEmbedded: """The safety net that surfaces produced inline figures the report forgot to embed.""" diff --git a/uv.lock b/uv.lock index 7c874c272..cb82b42c3 100644 --- a/uv.lock +++ b/uv.lock @@ -239,6 +239,9 @@ docs = [ { name = "sphinx-rtd-theme" }, { name = "sphinxcontrib-mermaid" }, ] +s3 = [ + { name = "boto3" }, +] viz = [ { name = "pygraphviz" }, ] @@ -270,6 +273,7 @@ requires-dist = [ { name = "aiofiles", specifier = ">=23.0.0" }, { name = "aiosqlite", specifier = ">=0.19.0" }, { name = "asyncpg", specifier = ">=0.29.0" }, + { name = "boto3", marker = "extra == 's3'", specifier = ">=1.35.0,<2" }, { name = "cryptography", specifier = ">=46.0.6,<47" }, { name = "deepagents", specifier = ">=0.6.5" }, { name = "knowledge-layer", extras = ["all"], editable = "sources/knowledge_layer" }, @@ -305,7 +309,7 @@ requires-dist = [ { name = "sphinxcontrib-mermaid", marker = "extra == 'docs'", specifier = ">=2.0.1" }, { name = "yapf", marker = "extra == 'dev'", specifier = ">=0.40.0" }, ] -provides-extras = ["dev", "docs", "viz"] +provides-extras = ["dev", "docs", "s3", "viz"] [package.metadata.requires-dev] dev = [