feat: add moss-connector-s3 — index an S3 bucket, re-index on change

[Sravan1011]

Summary

Adds an Amazon S3 source connector, built from the packages/moss-data-connector/_template pattern (with the DynamoDB connector as the boto3 style reference). It indexes documents from an S3 bucket and — per the watch-a-bucket story — re-indexes automatically when the bucket contents change.

Closes #381

What's included

S3Connector (src/connector.py)

• Lists the bucket with list_objects_v2, following continuation tokens automatically
• prefix filtering server-side, suffix filtering client-side (".md" or a tuple like (".md", ".txt")) — non-matching objects are skipped without being downloaded
• Zero-byte "folder" placeholder keys (ending in /) always skipped
• Hands the mapper a row dict: key, text (decoded body, configurable encoding), etag, last_modified, size, content_type, and S3 user metadata
• **boto3_kwargs passthrough (region_name, endpoint_url, credentials) so MinIO/LocalStack work out of the box
• snapshot() returns {key: etag} from a list-only pass — no bodies downloaded

watch() (src/watch.py) — the re-index-on-change piece

• Ingests once, then polls snapshot() on poll_interval and rebuilds the index whenever an object is added, removed, or modified
• Snapshot is taken before each ingest, so changes landing mid-ingest are caught on the next poll rather than lost
• max_polls for one-shot sync jobs and tests; on_change callback with the new snapshot
• Per the template rules, sync/loop logic lives in the connector's own code — ingest.py is copied verbatim from the template

Tests

• tests/test_s3.py — 12 unit tests, moto-mocked, no AWS needed: ingest end-to-end, prefix/suffix filters, pagination (page_size=1), empty bucket, auto_id, user-metadata passthrough, snapshot add/modify/delete tracking, watch re-index on change, and watch no-op when unchanged
• tests/test_integration_s3_moss.py — live round trip (S3 → ingest → Moss query → cleanup) plus a live watch test; skip-gated the same way as the DynamoDB connector (S3_ENDPOINT_URL for MinIO/LocalStack, or MOSS_CONNECTOR_S3_ALLOW_AWS=1 to opt in to real AWS)

Docs & extras

• Package README with one-shot ingest and watch-a-bucket usage, mapper row reference, MinIO/LocalStack notes
• demo.py end-to-end walkthrough (create bucket → upload → ingest → query → add object → watch re-indexes → cleanup)
• .env.example, and a new row in the parent moss-data-connector README layout + table

Test results

tests/test_s3.py ............ 12 passed
ruff check .                  All checks passed!

(Unit tests run fully offline via moto. Integration tests verified to collect and skip cleanly without credentials.)

[Copilot]

Pull request overview

Adds a new Python connector package under packages/moss-data-connector/ that can ingest documents from an Amazon S3 bucket into a Moss index, plus a polling-based watch() helper that re-indexes when bucket contents change (based on {key: etag} snapshots). This fits the existing moss-data-connector ecosystem by following the _template ingest pattern and matching the boto3 style used by the DynamoDB connector.

Changes:

• Added moss-connector-s3 package: S3Connector (list/get + filtering + snapshot), ingest(), and watch() re-index loop.
• Added unit tests (moto-backed) and live integration tests (skip-gated) for S3→Moss ingest + watch behavior.
• Added docs/demo assets and registered the connector in the parent moss-data-connector README.

Reviewed changes

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
packages/moss-data-connector/README.md	Registers the new S3 connector in the connector list/table.
packages/moss-data-connector/moss-connector-s3/src/init.py	Re-exports S3Connector, ingest, and watch as the package API.
packages/moss-data-connector/moss-connector-s3/src/connector.py	Implements S3 listing, filtering, object download/decoding, and snapshotting.
packages/moss-data-connector/moss-connector-s3/src/ingest.py	Template-style ingest helper that (re)creates a Moss index from a source iterable.
packages/moss-data-connector/moss-connector-s3/src/watch.py	Implements polling loop to detect bucket changes via snapshots and re-ingest.
packages/moss-data-connector/moss-connector-s3/tests/test_s3.py	Moto-based unit tests for filtering, pagination, snapshot diffing, and watch behavior.
packages/moss-data-connector/moss-connector-s3/tests/test_integration_s3_moss.py	Skip-gated live integration tests for end-to-end ingest/query and watch re-indexing.
packages/moss-data-connector/moss-connector-s3/README.md	End-user docs: usage, mapper schema, filtering, pagination, and MinIO/LocalStack notes.
packages/moss-data-connector/moss-connector-s3/pyproject.toml	Defines package metadata, deps (boto3/moss), and dev/test tooling config.
packages/moss-data-connector/moss-connector-s3/demo.py	Runnable demo script showing bucket setup → ingest → query → watch → cleanup.
packages/moss-data-connector/moss-connector-s3/.gitignore	Package-local ignores for venv/build/test caches and .env.
packages/moss-data-connector/moss-connector-s3/.env.example	Example env vars for S3 endpoint + Moss credentials.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[HarshaNalluru]

@cubic-dev-ai review this pull request

[cubic-dev-ai]

@cubic-dev-ai review this pull request

@HarshaNalluru I have started the AI code review. It will take a few minutes to complete.

[cubic-dev-ai]

All reported issues were addressed across 12 files

<sub>Tip: instead of fixing issues one by one fix them all with cubic

Re-trigger cubic</sub>

[HarshaNalluru]

High: watch() re-index calls create_index() on an index that already exists

watch() -> _sync(empty=False) -> ingest() -> MossClient.create_index(index_name, docs). The initial ingest creates the index, but every subsequent change re-enters the same path and calls create_index() again on a name that already exists (src/watch.py:846 initial, src/watch.py:856 on each change).

Evidence in this repo that create_index is create-only, not idempotent:

• packages/agno-moss/src/agno_moss/runtime.py:141-147 is the one first-party package that updates an existing index, and it explicitly branches: create_index only when not exists(), else add_docs(..., upsert=True). If create_index overwrote, that branch would not be needed.
• The SDK docstring reads "Create a new index" (sdks/python/sdk/src/moss/client/moss_client.py:85).
• No test or example anywhere in the repo calls create_index twice on the same name. Every SDK test uses generate_unique_index_name(...) and deletes on teardown.

Why the current tests do not catch it: FakeMossClient.create_index (tests/test_s3.py) just appends to a list and never errors on a duplicate name, so test_watch_reindexes_on_change asserting len(calls) == 2 passes regardless. The integration test would exercise the real backend but is skip-gated, and the reported test results only cover test_s3.py. The live watch path and demo.py step 5 appear not to have been run end to end. The core create_index lives in the compiled inferedge-moss-core crate, so this cannot be proven statically, but the convention evidence is strong enough that it should be verified before merge.

Same bug, second trigger: if watch() is restarted while the index already exists from a prior run, the initial _sync(empty=False) hits it immediately.

Suggested fix: make the rebuild path delete-then-create, mirroring the empty branch that already deletes and matching the documented "re-creates the index on every bucket change" intent:

async def _sync(..., *, empty: bool) -> None:
    client = MossClient(project_id, project_key)
    try:
        await client.delete_index(index_name)
    except Exception:
        pass  # not-found on first run / after an empty transition is fine
    if empty:
        return
    docs = await asyncio.to_thread(list, source)
    await ingest(docs, project_id, project_key, index_name, model_id=model_id, auto_id=auto_id)

Then add a test where the fake raises on duplicate-name create_index, and run the integration test once against MinIO plus a real Moss project to confirm.

Medium: full re-embed of the whole bucket on every change

Even after the fix above, rebuilding re-embeds every object whenever any single one changes. A one-file edit in a 10k-object bucket re-embeds all 10k, which is costly in time and embedding spend. The snapshot diff already knows exactly which keys were added, removed, or modified, so incremental add_docs plus delete_docs would be much cheaper. Full rebuild is a reasonable v1 for correctness since it is the only way to reflect deletes without diffing, but the tradeoff is worth calling out in the README watch section, with incremental sync as a follow-up. Not a blocker.

Medium: delete_index in the empty branch is not guarded

_sync's empty branch calls delete_index(index_name) unguarded (src/watch.py:886). agno-moss wraps every delete_index in try/except because it can raise when the index is absent. In watch() the empty branch only fires on a non-empty to empty transition, so the index should exist, but a concurrent external delete would then crash the watcher. Wrap it defensively. The suggested fix above centralizes this anyway.

Low / nits

• ingest.py is byte-identical to _template/src/ingest.py, consistent with the keep-in-sync rule. Good.
• The parent README already omits moss-connector-huggingface. Since this PR edits that exact table, adding the missing huggingface row would be a cheap cleanup. Optional.
• _client() builds a fresh boto3.client("s3") on every snapshot() and every iter (src/connector.py:669). For a 60s-interval watcher this is negligible and matches the DynamoDB connector's per-iter resource creation, so it is convention-consistent. Caching on the instance would trim minor overhead.
• page_size default 1000 vs DynamoDB's 100 is correct, since 1000 is the S3 list_objects_v2 maximum, and it is documented.

[Sravan1011]

@HarshaNalluru done with the changes

[HarshaNalluru]

@Sravan1011 please resolve the conversations.

[Sravan1011]

@Sravan1011 please resolve the conversations.

I have resolved all, can you specify which one ? I have also made change for the latest comment also

[HarshaNalluru]

@Sravan1011 go to each of the comments to check which ones are unaddressed.

[github-actions]

Codex review

No issues found.