Data Pipeline

What the source does

• FilesystemNone observedNo reads or writes in the source
• NetworkNone observedNo outbound calls in the source
• ShellNone observedNo commands run in the source
• DependenciesPinnedNothing floats
• SecretsNone foundNo keys in source

Five audit layers

Gates applied: no behavioural pass, no license.

Audit 0.4.1 · source sha dd4da722e22a · 18 Oct 2026Report an issue or request a re-scan

• Publisheragulli GitHub userOwner of the repository this skill is read from
• Source folderagulli/atlas-agents/ch09_agent_skills/skills/data-pipelineThe exact folder we audited
• LicenceNot declaredAs the repository states it
• Commit pinned for this verdictdd4da722e22aA new audit runs when the source changes
• Repository statusNot archivedLast push 93 days ago
• Sandboxed runNot runSkills are read, not executed; layer L4 is skipped

Its description: Design, build, or debug data processing pipelines. Use when asked to process a dataset, transform data, build an ETL pipeline, schedule batch jobs, or fix data quality issues.

---
name: data-pipeline
description: Design, build, or debug data processing pipelines. Use when asked to process a dataset, transform data, build an ETL pipeline, schedule batch jobs, or fix data quality issues.
license: MIT
compatibility: Requires python 3.10+
---

## Overview

Data pipelines fail silently and corrupt downstream systems. Every pipeline must be observable, idempotent, and validated at the boundary.

## Process

1. **Define the contract.** Before writing any transformation code, specify:
   - **Input schema:** What fields, types, and constraints does the data arrive with?
   - **Output schema:** What fields, types, and constraints must the output satisfy?
   - **Volume:** How many records? Per-run? Per-day?
   - **Frequency:** One-time, scheduled, or event-driven?

2. **Validate at the boundary.** The first thing any pipeline stage does is validate its input:
   ```python
   from pydantic import BaseModel, ValidationError

   class InputRecord(BaseModel):
       user_id: int
       event_type: str
       timestamp: str  # ISO 8601
       value: float | None = None

   def process(raw_records: list[dict]) -> list[dict]:
       valid, invalid = [], []
       for r in raw_records:
           try:
               valid.append(InputRecord(**r).model_dump())
           except ValidationError as e:
               invalid.append({"record": r, "error": str(e)})
       if invalid:
           log_invalid_records(invalid)  # Never silently drop
       return transform(valid)
   ```

3. **Make it idempotent.** Running the pipeline twice on the same input must produce the same output. Use upserts, not inserts. Use deterministic IDs based on input content, not auto-increment.

4. **Log progress at meaningful checkpoints.** After every major stage (extract, validate, transform, load), log the record count and any failures.

5. **Test with a sample.** Before running on the full dataset, run on 100 records. Confirm the output schema, record count, and that no records were silently dropped.

6. **Run on the full dataset.** Monitor progress. On completion, report: records in, records out, records failed, and time elapsed.

## Rationalizations

| Excuse | Rebuttal |
|--------|----------|
| "I'll add validation later" | Invalid data corrupts your database. Validate at the boundary now. |
| "Logging slows the pipeline down" | A pipeline that fails without logs requires a full rerun to debug. Log it. |
| "It worked on the sample" | Test samples are not representative. Always run a full-dataset dry run before writing to the destination. |

## Verification

- [ ] Input and output schemas are defined before any code is written
- [ ] Invalid records are logged, not silently dropped
- [ ] Pipeline was tested on a 100-record sample before full run
- [ ] Final report includes: records in, records out, records failed

Excerpt, first 2,852 characters of SKILL.md, verbatim from the audited commit. It is text from someone else’s repository, quoted so you can judge it before your agent reads it.

git clone https://github.com/agulli/atlas-agents
git -C atlas-agents checkout dd4da722e22a

Works in

1. 2026-10-18
   ✓SAFE B 89 / 100

   first audit · source dd4da722e22a

Questions about Data Pipeline

• github.com/agulli/atlas-agents/tree/main/ch09_agent_skills/skill — the audited source at commit dd4da722e22a and every finding aboveObserved · 18 Oct 2026

Spotted something wrong? Report a correction · Methodology