Debug Pipeline -- Troubleshooting Guide

Investigate the pipeline issue described via $ARGUMENTS.

Follow the diagnostic steps below. Start with Dagster MCP tools to gather concrete error data, then investigate the code.

Step 1: Triage via Dagster MCP

Use the Dagster MCP tools to gather diagnostic data before reading code. These tools query the running Dagster instance directly.

1a. Discover What's Deployed

Use mcp__dagster__list_repositories to see all code locations. Then use mcp__dagster__list_assets and mcp__dagster__list_jobs with the repository location and name to understand what's running.

1b. Find Recent Failures

Use mcp__dagster__recent_runs (default limit: 10) to find failed runs. Look for runs with FAILURE status.

If the user hasn't provided a run ID, show them the recent failures and ask which one to investigate. If only one failure exists, proceed with that.

1c. Get Detailed Error Info

Use mcp__dagster__get_run_info with the run ID to get:

• Run status and timestamps
• Error messages and stack traces
• Step-level failure details
• Run configuration

This replaces manually checking the Dagster UI. The error message from get_run_info is usually sufficient to identify the root cause.

1d. Check Asset Status

Use mcp__dagster__get_asset_info with the asset key to check:

• Latest materialization status
• Partition status
• Upstream dependency state

If the user mentions a specific asset but not a run ID, start here. Ask the user for the asset key if it's not clear from context.

When to Ask the User

Ask the user for identifying information when you need it:

• Run ID: "Which run failed? I can see these recent failures: [list from recent_runs]"
• Asset key: "Which asset is failing? Here are the assets in this repository: [list from list_assets]"
• Repository: Only ask if multiple code locations exist (rare in dev)

Step 2: Diagnose by Symptom

Use the error information from Step 1 to match against these known patterns.

Asset Won't Materialize

Possible causes:

1. Automation condition not triggered: Check that upstream asset has a version change

   • Read: packages/pipeline/swiss_ai_hub/pipeline/automation/all_deps_completed.py
   • Verify asset has AutomationCondition.eager() or all_deps_completed

2. Automation sensor not running: The sensor must be included in Definitions

   • Read: packages/pipeline/swiss_ai_hub/pipeline/sensors/factory.py
   • Verify default_automation_sensor(assets) is in the sensors list

3. Partition key mismatch: Dynamic partitions may not have been created

   • Read: packages/pipeline/swiss_ai_hub/pipeline/util/partition_utils.py
   • Check max_partitions limit -- if exceeded, some partitions are deferred

4. Resource missing: Op can't find required resource

   • Check the resources dict in the Definitions matches all ResourceParam[T] types in ops
   • Stage 2 always supplies every enrichment resource (table_refinement, summary_parser): the graph is the same for every knowledge database, and each step decides per run from the bucket's configuration whether it has work. A missing one is a wiring bug, never a deployment setting.

5. IO Manager mismatch: Asset output type doesn't match IO manager expectations

   • S3DataLakeIOManager expects DataLakeFile
   • DocStoreIOManager expects RefDocDocument
   • VectorStoreIOManager expects list[TextNode]

Observable Asset Returns No Changes

Possible causes:

1. No files in source: The data lake / rclone remote is empty

   • Check S3 bucket contents
   • For rclone: verify remote is configured and accessible

2. Hash unchanged: Content hash matches previous observation

   • Data versions are content-based -- if file content didn't change, no version change
   • Read: packages/pipeline/swiss_ai_hub/pipeline/ops/data_lake/data_version_by_partition_for_data_lake.py

3. Partition limit reached: max_partitions caps how many partitions are processed per run

   • Default: 1000 partitions per observation
   • If more files exist, they'll be picked up in subsequent runs

Sensor Not Firing

Check automation sensor:

• Read: packages/pipeline/swiss_ai_hub/pipeline/sensors/factory.py
• Is default_automation_sensor(assets) in the Definitions.sensors list?
• Is minimum_interval_seconds appropriate? (default: 60s)

Check NATS sensor:

• Read: packages/pipeline/swiss_ai_hub/pipeline/sensors/nats/nats_document_uploaded_sensor.py
• Is NATS reachable? Check NATS_ENDPOINT env var
• Is the JetStream stream created? (poller creates it on first run) — events published before the stream exists are dropped, since JetStream has nowhere to store them
• Are SourceUpdatedEvent messages being published?
• Check consumer name uniqueness
• skipped: Observation run <id> is already in flight is normal, not a fault — the sensor deliberately keeps one observation at a time and re-arms afterwards. Likewise Debouncing N pending event(s) means a burst is being collected (30 s). Read the sensor's cursor in the Dagster UI to see pending_events / followup_armed; if a run seems missing, check whether its dagster/run_key was deduplicated

Embedding Failures

Symptoms: RetryPolicy exhausted, ValidationError in embed_nodes

• Read: packages/pipeline/swiss_ai_hub/pipeline/ops/nodes/embed_nodes.py
• LiteLLM endpoint unreachable
• Token limit exceeded: document too large after chunking
• Batch splitting: on ValidationError, batch is recursively split in half
• After 6 retries with exponential backoff, op fails permanently

Document Parsing Failures

Symptoms: parse_document_from_data_lake fails

• Read: packages/pipeline/swiss_ai_hub/pipeline/ops/data_lake/parse_document_from_data_lake.py
• Read: packages/pipeline/swiss_ai_hub/pipeline/resources/parser/document_parser_resource.py
• MinerU service not running (if using remote MinerU)
• Unsupported file type
• Corrupted file content
• Large file timeout
• MinerU and MarkItDown conversions are cached by content hash in the parse-cache bucket (MineruParseCache, MarkItDownParseCache, both in packages/core/swiss_ai_hub/core/generative_ai/document/loaders/); a stale or wrong result for the same bytes means the cache entry (keyed also by MinerU settings, or by the MarkItDown version and XlsxMarkdownConverter's format) is the thing to inspect or delete, entries expire by lifecycle rule

Step 3: Check Infrastructure Connections

S3 (SeaweedFS)

Symptoms: botocore.exceptions.ClientError, EndpointConnectionError

Settings: packages/core/swiss_ai_hub/core/infrastructure/s3/s3_storage_settings.py

Common issues: SeaweedFS not running (docker compose -f infra/docker-compose.dev.yml ps seaweedfs-s3), bucket doesn't exist, wrong endpoint (inside Docker use http://seaweedfs-s3:9000, not localhost).

Milvus

Symptoms: MilvusException, connection timeout

Settings: packages/core/swiss_ai_hub/core/infrastructure/milvus/milvus_settings.py

Common issues: Milvus not running, dimension mismatch (embedding model dimensions must match collection dimensions), collection not found (auto-created on first insert).

MongoDB/FerretDB

Symptoms: ConnectionFailure, OperationFailure

Settings: packages/core/swiss_ai_hub/core/infrastructure/mongo/mongo_settings.py

Common issues: FerretDB not running, database name mismatch (derived from bucket name via get_db_name_from_bucket_name() in packages/pipeline/swiss_ai_hub/pipeline/util/bucket_utils.py).

Rclone

Symptoms: ClientConnectorError, 403 Forbidden, remote not found

Settings: packages/core/swiss_ai_hub/core/infrastructure/rclone/rclone_settings.py Client: packages/pipeline/swiss_ai_hub/pipeline/resources/rclone/rclone_client.py Per-run remote: packages/pipeline/swiss_ai_hub/pipeline/util/source_builders.py

Common issues: Rclone service not running, the daemon lost its remotes after a restart (expected, it keeps no config file: the next run re-creates them from the database's stored source configuration), OAuth token expired, wrong --rc-serve flag (required for file downloads), a config/create error such as invalid key or value contains \n (a multi-line credential; Google Drive JSON and SFTP PEM keys are normalised, other backends are not).

NATS

Settings: packages/core/swiss_ai_hub/core/infrastructure/nats/nats_settings.py

Common issues: NATS not running, JetStream not enabled, stream/consumer not created.

Docker Quick Check

# Check which services are running
docker compose -f infra/docker-compose.dev.yml ps

# Check specific service logs
docker compose -f infra/docker-compose.dev.yml logs <service> --tail=50

Step 4: Common Error Patterns

"No partition definition found for the upstream asset"

IO manager can't find partition info. Check asset has partitions_def set and IO manager handles both partitioned and non-partitioned cases.

"Expected a DataLakeFile or a list of DataLakeFiles"

S3DataLakeIOManager received wrong type. Check asset's io_manager_key matches the correct IO manager and op return type matches expectations.

"Cannot connect to host seaweedfs-s3:9000"

S3 unreachable. Check the SeaweedFS S3 gateway is running, S3_STORAGE_ENDPOINT is correct, network connectivity between Dagster and SeaweedFS containers.

"No nodes found for document after retrying for 30 seconds"

VectorStoreIOManager eventual consistency timeout. Check Milvus is healthy, nodes were actually inserted (check collection), and document ID conversion (URI to hash) is consistent.

"Rclone remote not found"

Read: packages/pipeline/swiss_ai_hub/pipeline/util/source_builders.py (rclone_remote_for_bucket).

Remotes are named rclone_{bucket} and rebuilt from BucketEntity.source_configuration on every run; nothing is configured by hand. A missing remote therefore means the run never reached config/create: the database has no source any more, its stored configuration fails validation (check the run's first step), or the daemon at RCLONE_URL is not the one the pipeline talks to.

Step 5: Re-trigger After Fix

Once the root cause is fixed, use MCP tools to verify the fix:

• Re-materialize an asset: Use mcp__dagster__materialize_asset with the asset key, repository location, and repository name
• Re-launch a job: Use mcp__dagster__launch_run with the job name and repository info
• Verify success: Use mcp__dagster__recent_runs and mcp__dagster__get_run_info to confirm the re-run succeeded
• Terminate stuck run: Use mcp__dagster__terminate_run if a run is hanging