# Obtainer

> Use this skill when LoopAI needs dataset discovery, acquisition, web-page collection, DataMixer lakehouse operations, data processing, indexing, recipe planning, or production training-data export. In long-running Codex SDK loops, when Analyzer produces an analysis report, failure taxonomy, or user request that implies new training data is needed, Codex must activate this Obtainer skill, parse the data need into an intent, and delegate the whole workflow to the Obtainer Orchestrator agent (`dm obtainer-orchestrator start`), then poll its structured status. The orchestrator owns lake bootstrap and the dispatch/gating of the managed sub-agents (dataset-acquisition-agent, dataflow agent-run, sft-export-agent); the outer Codex context must not run lake init, acquisition bridges, download manifest, ingest, or export itself for a normal obtain task. The rest of this skill is the domain policy the orchestrator worker follows.

- Skill: `opendcai/obtainer` (Agent Skill)
- Install (CLI): `npx skillmds@latest add opendcai/obtainer`
- Raw SKILL.md: https://api.skillmd.com/api/skills/opendcai/obtainer/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Product & Planning
- Author: opendcai (https://skillmd.com/u/opendcai)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/opendcai/obtainer

---


# Obtainer Skill

## Purpose

Obtainer is the agent-facing workflow for turning a data need into a production
training-data artifact. SearchAgent discovers hosted datasets, while the
registered Domain Data Acquisition WebAgent (`domain_data_acquisition`, legacy
alias `webcrawler_dm`) collects primary vertical-domain webpages as raw L1 data. DataMixer
is the only data-lake command surface for storage, ingest, processing,
indexing, sampling, recipe planning, export, snapshots, and lineage.

ObtainerCLI is the only supported end-to-end data workflow. Requests to clean,
deduplicate, quality-filter, map, construct, or export a training dataset are
Obtainer requests and must stay in the ObtainerCLI/DataMixer workflow through
the final artifact.

When a long-running Codex SDK loop receives an Analyzer report, failure taxonomy,
training recipe, or next-iteration data request, treat it as an Obtainer input,
not a generic coding task:

1. Identify whether the report needs dataset acquisition, production export, or both.
2. For acquisition/download/ingest, start the managed
   `dataset-acquisition-agent` worker instead of manually driving
   SearchAgent/WebAgent/download/ingest from the outer Codex context.
3. Poll worker status and decide whether to resume the same worker or start a
   fresh worker.
4. Run the mandatory DataFlowAgent post-processing stage (`dm dataflow
   agent-run`), which materializes the L4 dataset (quality, decontamination,
   deduplication, normalization, safety, and post-training validity).
5. Only after the DataFlowAgent run completes and the final L4 dataset scale
   meets the recipe target, start the managed `sft-export-agent` worker for
   production SFT outflow. If the user explicitly specifies an L3 export, the
   L4 gate is waived and L3 data may be exported directly once the lake
   volume/mix/quality gates pass.
6. Report warehouse path, datasets, record counts, recipe/export artifacts,
   lineage, manifests, and snapshots.



## Main-Agent Use: Delegate to the Obtainer Orchestrator

The main agent (starter) must NOT bootstrap the lake, dispatch acquisition /
export workers, or run DataFlowAgent itself for a normal obtain task. Those
responsibilities belong to the dedicated Obtainer Orchestrator agent
(`dm obtainer-orchestrator`). The orchestrator owns lake bootstrap, sub-agent
dispatch, progress gating and the final deliverable report; the policy below is
its domain policy. The main agent only parses intent, starts the orchestrator,
polls its structured status, and reports the terminal artifacts.

### 1. Parse the data need into an intent

Extract from the Analyzer report / user request / recipe: an `--objective`
(what sample shape is needed), `--keywords` (search / domain hints),
`--target-datasets` (how many buckets/datasets), and a compact `--message`
(failure taxonomy, quality gates, proportions).

### 2. Start the orchestrator

```bash
${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm \
  obtainer-orchestrator start \
  --run ./outputs/obtainer_run_<timestamp> \
  --objective "buggy and fixed Python code pairs for syntax repair SFT" \
  --keywords "python syntax error, code repair dataset" \
  --target-datasets 2 \
  --message "Analyzer report: ...; require license=unknown and quality>=0.8" \
  --python-executable /path/to/loopai-env/bin/python
```

`start` launches the orchestrator's inner Codex SDK worker in the background
and returns the run directory. Use `--foreground` only when you intend to
block.

### 3. Poll the orchestrator (fine-grained status contract)

A full obtainer orchestration runs for roughly **3-4 hours** (acquisition +
DataFlow L4 + export). **You MUST NOT poll more often than every 5 minutes.**
Between polls run `sleep 300 && ... status ...`; polling faster wastes tokens and
does not speed up the run. `updated_at` / `stale` in the status tell you whether
the orchestrator is alive far better than polling frequency does:

```bash
${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm \
  obtainer-orchestrator status --run ./outputs/obtainer_run_<timestamp> --json
```

Read the machine-readable contract (`schema_version: 1`):

- `state`: `idle | running | completed | completed_with_errors | failed | interrupted | stopped`
- `phase`: `bootstrap | acquiring | gating | dataflow | exporting | finalizing`
- `progress` (0..1), `message`, `updated_at` (heartbeat), `stale`
- `next_action`: `poll` -> keep polling; `start_dataflow` -> the lake volume
  gate already passed while acquisition is still running - the orchestrator
  should dispatch DataFlow L4 in parallel, keep polling; `report` -> read
  final_report.json and report; `resume` -> the orchestrator concluded while
  sub-agents were still running or returned no valid result, run `resume` to
  continue; `blocked` -> surface error + gates to the user
- `subtasks[]`: each managed sub-agent (state / progress / message / run_dir)
- `gates[]`: e.g. `lake_volume`, `dataflow_l4` with `ok` + `detail`
- `lake`: warehouse, dataset / record counts, quality_levels

Never judge progress from `message` alone; use the structured fields.

### 4. Terminal handling

- `completed` / `next_action=report`: read `final_report.json` in the run dir
  and report warehouse, datasets, record counts, recipe / export artifacts,
  lineage, manifests and snapshots.
- `interrupted` / `next_action=resume`: the orchestrator concluded while a
  sub-agent was still running or returned no valid result - run
  `${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm \
  obtainer-orchestrator resume --run <dir> --message "<why / resume from where>"`,
  do NOT take over its sub-agents.
- `failed` / `next_action=blocked`: read `error` + failing `gates`, tell the
  user, and offer `resume` once the blocker is addressed:
  `${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm \
  obtainer-orchestrator resume --run <dir> --message "<why / resume from where>"`
- `stale=true` while `state=running`: warn that the orchestrator may be hung and
  offer `stop` or `resume`.

### Hard constraints for the main agent

- Never run `dm lake ...`, `dataset-acquisition-agent`, `sft-export-agent`,
  `dataflow agent-run`, `searchagent`, `webagent` or `download manifest`
  yourself for a normal obtain task - the orchestrator owns those.
- **Never `kill` / `pkill` the orchestrator's worker processes.** The
  orchestrator worker is managed by the CLI (`start` / `resume` / `stop`); raw
  process kills leave it in a stuck `running` state and break the run. If the
  status looks stuck, first check `updated_at` / `stale`; only then use
  `dm obtainer-orchestrator stop --run <dir>` followed by `resume` (never raw
  `kill`), and keep polling otherwise.
- Never claim obtainer completion without a `final_report.json` reported by the
  orchestrator.

## Hard Constraints

- **DataMixer-only lakehouse.** Do not use non-DataMixer lake logic, standalone
  table sampling, compatibility shims, or hand-written tiny fixtures for lake
  operations. If a DataMixer command cannot satisfy the request, stop and report
  the blocker.
- **Outer Codex must delegate acquisition.** For any normal dataset discovery,
  download, normalization, or ingest request, the outer Codex context must start
  the CLI wrapper `loopai-obtainercli dm ... dataset-acquisition-agent start`
  or run `${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm ... dataset-acquisition-agent
  start`. If the outer shell is not using the LoopAI environment, set
  `LOOPAI_PYTHON_EXECUTABLE=/path/to/loopai-env/bin/python` or pass
  `--python-executable /path/to/loopai-env/bin/python`, then poll/resume that worker.
  Do not use a generic `spawn_agent`
  worker for data acquisition. Do not create a SearchAgent task JSON, call
  `searchagent`, call `download manifest`, normalize files, or ingest rows from
  the outer Codex context. Those operations belong inside the CLI worker policy.
- **DataMixer is the only lake command surface.** Use
  `loopai-obtainercli dm ...` for initialization, schema inspection, dataset
  registry, ingest, query, processing operators, indexing, recall, recipes,
  snapshots, lineage, and export.
- **Reuse the active DataMixer warehouse.** Treat `.datamixer/lake.yaml` as a
  project pointer to a reusable DataMixer warehouse. Do not create a new lake per
  task unless the user explicitly asks for a new warehouse. Use `dm lake load`
  to point the project at an existing warehouse and `dm lake delete` to unload
  the pointer; deletion preserves the warehouse unless `--delete-warehouse
  --yes` is explicitly supplied. Prefer `dm lake scan` before choosing a
  warehouse, so the agent sees project and cache candidates instead of guessing
  paths.
- **Use lake context, not repeated boilerplate.** After a lake is loaded or
  initialized, use `dm --lake .datamixer/lake.yaml ...` for agents. The pointer
  persists the warehouse, selected WebAgent, model name, worker/subquery
  defaults, current acquisition run, and current campaign id. Do not pass a
  FastAPI/Configer SQLite file as `--root`; `--root` must be a DataMixer
  warehouse containing `datamixer.toml`.
- **Load or init the lake before any worker.** `dataset-acquisition-agent` and
  `sft-export-agent` refuse to start (`LAKE_NOT_LOADED`) unless the resolved
  warehouse already contains `datamixer.toml`. When a previous task ended, clear
  its stale bindings first with `dm lake unbind` so the pointer never confuses
  the new run with an old task_id; then start the worker with
  `dm --lake .datamixer/lake.yaml ...`.
- **Prepare worker intent before acquiring from a report.** First recognize the
  dataset-acquisition intent: target sample shape, task types, domains, source
  hints, proportions, quality gates, and concrete search objectives. Pass that
  intent to `dataset-acquisition-agent start` via `--objective`, `--keywords`,
  `--target-datasets`, and `--message`. The worker may then use SearchAgent
  internally. Never pass the raw Analyzer report as the only search target.
- **Objectives describe dataset shape, not only error keywords.** Use objectives
  like "buggy and fixed Python code pairs for syntax error repair", not only
  "SyntaxError" or "missing".
- **Continuous dual-stream pipeline:** inside the acquisition worker, start
  SearchAgent and the registered `domain_data_acquisition` campaign concurrently.
  It is a vertical-domain data source collector, not a general browser helper.
  SearchAgent finds hosted datasets for the provider download manifest; WebAgent
  collects primary webpages into a distinct DataMixer L1 dataset. Start WebAgent
  detached with its L1 -> L2 -> L3 streaming pipeline enabled. Its downstream
  queues consume new L1 rows while collection continues. Wait only for the
  SearchAgent artifact needed for hosted downloads; do not wait for the WebAgent
  campaign to complete before filtering, downloading, normalizing, ingesting, or
  beginning the next planned DataMixer stage. Retain separate artifacts/statuses
  in `final_report.json`. A launch or persistent processing failure is terminal;
  an active WebAgent campaign is not.
- **Lake readiness is the downstream gate:** while acquisition continues, poll
  per-bucket record/token counts and the planned quality gates. As soon as the
  lake satisfies the required volume, mix, and quality, immediately start the
  DataFlowAgent post-processing stage and required indexing and recipe planning;
  start production export only after the L4 gate below passes (or directly,
  when the user explicitly specifies an L3 export). Never use
  WebAgent completion, acquisition-worker completion, or empty producer queues
  as prerequisites; keep those producers running concurrently.
- **DataFlowAgent is a mandatory pre-export gate by default.** Every
  production export must first complete the DataFlowAgent post-processing stage
  (`dm dataflow agent-run`), which delivers a trial-verified L4 pipeline; the
  outer Codex then executes it over the exported 1.5x bucket-buffer input with
  the chunked runner to produce the L4 dataset. L4 is the DataFlow-processed
  level on top of the L1 -> L2 -> L3 chain (raw webpages -> normalized PT ->
  SFT QA -> post-processed) and is the default sample source for production
  export. Skipping, deferring, or folding this stage into the export worker is
  not allowed; an export without a completed L4 source is a blocker. If the
  user explicitly specifies an L3 export, the L4 gate is waived and L3 data
  may be exported directly instead.
- **1.5x in-lake redundancy is a hard requirement.** To guarantee that the
  export mix can be met, the lake must hold at least 1.5x the recipe target
  volume both overall and per bucket (`available_samples >= 1.5 x
  target_samples` for every bucket). If any bucket falls below this floor,
  continue acquisition and DataFlow post-processing until the redundancy is
  satisfied; never export a mix from a non-redundant lake.
- **L4 scale gates export by default.** The DataFlowAgent stage is considered
  complete only when the final L4 dataset scale meets the recipe target
  (overall and per bucket, after the 1.5x redundancy floor). Only then call
  `sft-export-agent`. If the user explicitly specifies an L3 export, this L4
  scale gate is waived and L3 data may be exported once the lake
  volume/mix/quality gates pass.
- **WebAgent model prerequisite:** the wrapper resolves the Codex default model,
  registers that same provider in the DataMixer model pool, and records
  `resolved_model`, `webagent_model`, and `model_source` in `thread.json`.
  WebAgent must use that exact `webagent_model`; never select an arbitrary local
  model from `dm model list` and never continue when the value is absent.
- **Worker must inspect `searchagent_manifest.json` before downloading.** If
  errors are non-empty, the download list is empty, candidates are unrelated to
  the interpreted intent, or sources cannot satisfy the requested sample shape,
  the worker refines the search once. If still unsuitable, stop and report the
  mismatch.
- **Worker must prune unrelated candidates before download.** After internal
  SearchAgent returns a download list, the worker compares every candidate
  against the original user request and interpreted dataset intent. Remove
  datasets that are clearly unrelated in domain, task type, language, source
  family, target label shape, or training purpose before the worker runs
  `download manifest`. Write a filtered manifest and a rejection list with
  explicit reasons; do not download the raw manifest when it contains unrelated
  candidates.
- **Worker stops on download failure.** If internal `download manifest` fails,
  is interrupted, or creates partial/empty files for selected datasets, stop
  before ingest. Report the command, exit code, produced files, and blocker.
- **Acquisition download cap.** Internal `download manifest` writes at most
  100,000 rows and 2GiB of local JSONL output per dataset, even if `--max-rows
  0`, a larger row value, or an oversized `--max-bytes-per-dataset` value is
  supplied. If the byte cap is reached, keep the partial JSONL and report
  `truncated`, `truncated_reason`, `rows_written`, and `bytes_written`. Treat
  this as the bounded acquisition bridge into DataMixer, not as final
  production SFT output.
- **Production SFT budget.** If the Analyzer report or user gives no explicit
  SFT target, set and report a production default before export: at least
  100,000 total records, or an explicit token budget when token counts are
  available.
- **Plan recipe proportions from the current need.** Do not assume a fixed
  bucket mix from examples or prior runs. The worker must choose and justify
  bucket proportions from the current user goal, Analyzer failure taxonomy,
  available lake inventory, quality filters, and record/token budget. For
  token-budget recipes, allocate against `total_tokens`; for sample-budget
  recipes, allocate against `total_samples`. Persist an acquisition
  `manifest/data_mix_plan.json` before discovery and an export
  `recipe/recipe_plan.json` plus `recipe/mix_plan.json` before outflow. The
  managed workers reject successful completion when those artifacts are absent,
  inconsistent, or lack per-bucket rationale.
- **Use semantic recipe filters.** Failure-taxonomy exports must use meaningful
  tags or columns such as `bug_type=syntax`, `bug_type=logic`,
  `bug_type=runtime`, and `bug_type=assertion`. If those tags do not exist in
  enough volume, stop and report that the lake cannot guarantee the requested
  mix. Do not replace them with broad proxies such as only `lang=python`.
- **Complete metadata on ingest.** Preserve source platform, source dataset
  id/name, source URI, license, language, domain, task type, processing level,
  source kind, split, loop UUID, and version id. Unknown values must be explicit,
  for example `license=unknown`; do not silently omit required provenance.
- **Two lake paths, one quality model.** WebAgent ingests per item through its
  L1 -> L2 -> L3 pipeline, where `domain_classify` judges each row against the
  campaign's `--focus-keywords` and `topic_quality_filter` approves or rejects
  it. The dataset acquisition path is batch: `dm ingest` writes every
  normalized row with the batch metadata and an explicit `--quality-level`; it
  does not run per-record LLM approval and never drops rows. A dataset name,
  source name, URL, or ingest flag is not domain evidence, so a `--domain`
  flag is only batch-level metadata, never a per-row attestation.
- **Dataset cards and additive derivation on ingest.** For every acquired
  dataset, the acquisition worker must write and register a Markdown dataset
  card describing source, license, split, row count, original fields, derived
  fields, derivation rules, validation checks, intended training use, and known
  risks. Dataset-specific derived fields are allowed for embedded complex
  formats such as step traces, multi-turn conversations, or question+options,
  but derivation must be additive: preserve every original field, keep the row
  count unchanged, and validate that every declared derived field is non-empty
  before ingest succeeds.
- **Never overwrite or hide provenance.** Keep dataset lineage, loop/version
  tags, recipe fingerprints, export manifests, and snapshots.
- **Register user-named benchmarks before acquisition/ingest.** If the user
  query or Analyzer report explicitly names the benchmark type to collect (for
  example "collect HumanEval-style code problems", "BIRD SQL pairs", or any
  named eval/test set), treat that dataset as evaluation-only and register it
  in the DataMixer benchmark registration layer (`dm contam add --name
  <benchmark> --file <file>`) before any acquisition or ingest, so downstream
  ingest and export decontamination exclude it and benchmark data cannot leak
  into training data. Do not rely on a later `decontaminate` pass alone to
  decide what to register.

## Command Surface

Obtainer has one production data-lake command surface:

```bash
loopai-obtainercli dm --root /path/to/datamixer-warehouse <datamixer-command> --json
loopai-obtainercli dm --lake .datamixer/lake.yaml <datamixer-command> --json
```

Use `--root` when operating directly on a DataMixer warehouse. Use `--lake` only
when a LoopAI lake pointer already exists and should resolve to the integrated
DataMixer warehouse. All `dm` commands emit machine-readable JSON.

Manage the project pointer to a reusable DataMixer warehouse:

```bash
loopai-obtainercli dm lake scan --link .datamixer/lake.yaml --project-root .
loopai-obtainercli dm lake current --link .datamixer/lake.yaml
loopai-obtainercli dm lake load --warehouse /path/to/warehouse --link .datamixer/lake.yaml
loopai-obtainercli dm lake delete --link .datamixer/lake.yaml
loopai-obtainercli dm lake context --link .datamixer/lake.yaml
loopai-obtainercli dm lake unbind --link .datamixer/lake.yaml
```

`dm lake delete` unloads only the pointer by default. Use
`--delete-warehouse --yes` only when the actual reusable warehouse should be
removed.

SearchAgent, WebAgent, and provider download are internal acquisition bridges.
In the normal product workflow, outer Codex reaches them only by starting
`dataset-acquisition-agent`. Do not call low-level `searchagent`, `webagent`,
or `download manifest` from the outer Codex context.

## Dataset Acquisition Worker

For dataset discovery, WebAgent collection, candidate pruning, download,
normalization, and DataMixer ingest, outer Codex must use the managed acquisition
worker CLI wrapper. Here
"worker" means the `dataset-acquisition-agent start` command below, not a
generic spawned Codex worker.

Start a new worker:

```bash
${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm --lake .datamixer/lake.yaml dataset-acquisition-agent start \
  --run ./outputs/acquisition_run \
  --analysis-report ./outputs/analyzer_report.md \
  --objective "collect general-domain instruction and QA datasets" \
  --keywords "instruction tuning dataset, open QA dataset, summarization dataset" \
  --target-datasets 30 \
  --max-rows-per-dataset 100000 \
  --max-bytes-per-dataset 2147483648 \
  --discovery-mode auto \
  --python-executable /path/to/loopai-env/bin/python
```

`start` runs the inner Codex SDK worker in the background by default and returns
PID plus log paths. Use `--foreground` only when the caller intentionally wants
to block. If `loopai-obtainercli` is not installed as a console script, use the
`${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli ...` form.

Poll status:

```bash
${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm --lake .datamixer/lake.yaml dataset-acquisition-agent status \
  --run ./outputs/acquisition_run
```

Resume the same worker:

```bash
${LOOPAI_PYTHON_EXECUTABLE:-python} -m loopai.skills.ObtainerCLI.cli dm --lake .datamixer/lake.yaml dataset-acquisition-agent resume \
  --run ./outputs/acquisition_run \
  --message "Remove unrelated datasets from the filtered manifest, then continue ingest."
```

Do not pass `--model` to `dataset-acquisition-agent` unless the user explicitly
requests a one-off override. By default the wrapper resolves the Codex worker
model from Starter's model pool, preferring the configured Codex default model.

The worker wrapper injects the detailed acquisition policy: explicit objective
and keywords, concurrent SearchAgent/WebAgent discovery, candidate list review
against the original request before download, rejection report, 100,000-row and
2GiB JSONL-output per-dataset caps, normalized JSONL, DataMixer-only
ingest/status/query/index operations, complete provenance tags, and
`final_report.json`.

## DataMixer Lake Operations

Initialize and inspect:

```bash
loopai-obtainercli dm --root /path/to/warehouse init --json
loopai-obtainercli dm --root /path/to/warehouse status --json
loopai-obtainercli dm --root /path/to/warehouse schema --json
loopai-obtainercli dm --root /path/to/warehouse columns --json
loopai-obtainercli dm --root /path/to/warehouse stats --json
```

Dataset registry and ingest:

```bash
loopai-obtainercli dm --root /path/to/warehouse dataset add \
  --name code_repair_mix \
  --source huggingface \
  --license unknown \
  --description "buggy/fixed code repair datasets" \
  --json

loopai-obtainercli dm --root /path/to/warehouse ingest code_repair_mix \
  --file ./downloads/records/dataset.train.jsonl \
  --content-key content \
  --dataset-card ./manifest/dataset_cards/code_repair_mix.md \
  --derived-field train_output \
  --source-row-count 100000 \
  --stage sft \
  --domain code \
  --lang python \
  --source huggingface \
  --license unknown \
  --task-type SFT \
  --quality-level L3 \
  --tokenizer tiktoken:o200k_base \
  --json
```

If the downloaded file is not already normalized JSONL, use DataMixer
`agent-ingest`:

```bash
loopai-obtainercli dm --root /path/to/warehouse agent-ingest ./downloads/raw_file \
  --engine builtin \
  --dataset code_repair_mix \
  --quality-level L3 \
  --json
```

Query, coverage, and distributions:

```bash
loopai-obtainercli dm --root /path/to/warehouse query \
  --filter "domain = 'code' AND task_type = 'SFT'" \
  --limit 20 \
  --json

loopai-obtainercli dm --root /path/to/warehouse dist \
  --column domain \
  --json

loopai-obtainercli dm --root /path/to/warehouse grade \
  --filter "domain = 'code' AND task_type = 'SFT'" \
  --column quality_score \
  --json
```

Processing, quality, safety, and deletion:

When the user query or Analyzer report explicitly names a benchmark/eval
dataset type to collect, register it in the benchmark registration layer
first with `contam add` before any acquisition or ingest; the subsequent
`decontaminate` pass then excludes those rows from downstream training export
and prevents benchmark leakage.

```bash
loopai-obtainercli dm --root /path/to/warehouse op list --json
loopai-obtainercli dm --root /path/to/warehouse op run quality_score --dataset code_repair_mix --json
loopai-obtainercli dm --root /path/to/warehouse op run minhash_dedup --dataset code_repair_mix --arg k=5 --json
loopai-obtainercli dm --root /path/to/warehouse op run semantic_dedup --dataset code_repair_mix --json
loopai-obtainercli dm --root /path/to/warehouse contam add --name benchmark --file benchmark.txt --json
loopai-obtainercli dm --root /path/to/warehouse decontaminate --against benchmark --json
loopai-obtainercli dm --root /path/to/warehouse pii-redact --dataset code_repair_mix --dry-run --json
loopai-obtainercli dm --root /path/to/warehouse erase <sample_id> --reason "user request" --json
```

后处理阶段是必须要使用 dataflowagent 的，不要手工盲选单个 DataFlow operator。
`dataflow agent-run` 会让 Codex SDK 先导出试跑样本、按 DataFlow-Skills 规则
规划算子链、生成并试跑 pipeline；**试跑成功即交付**（`mode=trial_run`，
交付物 = `pipeline.py` + 试跑输出 `trial_processed.jsonl`）。**全量执行由
上层 Codex 负责**：拿到交付的 pipeline 后，用 chunk 脚手架跑
`full_input.jsonl`（1.5x 桶缓冲导出，不是全湖），产出 `full_processed.jsonl`
（L4），再按 `sample_id` 用 `apply-jsonl` merge 回 DataMixer。不要让
dataflowagent 自己跑全量或 merge。

**按桶 1.5x 缓冲导出，不是全量导出。** 调用 `agent-run` 时尽量带上出湖
`--recipe`（recipe.yaml）或 `--mix-plan`（mix_plan.json）：full input 会按
每个桶 `ceil(bucket_target * 1.5)` 行、固定 seed 抽样导出（可用行不足则全取），
避免把整个湖（动辄十几万行 / 数 GB）无谓地全量处理后处理。只对
`full_input.jsonl` 给到的行做后处理，不要自行重新全量导出或扩大范围。

**质量评估必须使用 DataFlow 的 LLM 评估算子**（如 `PromptedEvaluator` /
`PromptedFilter` 这类 LLM 打分/过滤算子），不得因耗时或成本而退化成纯启发式
规则打分；只有任务本身没有 LLM 打分语义、或 LLM serving 不可用时才允许规则
算子兜底并说明具体原因。不得覆盖原始字段和值；后训练内容需要构造或改写时，
使用生成算子写入新的派生字段，再使用 LLM 评估算子打分和筛选生成内容。

全量执行由上层用 chunked runner 跑，**可能非常耗时**——LLM 质量评估算子
逐条打分时，数小时到十几小时属正常，跑完为止。**不要用外层 shell `timeout`
包住 agent-run 或 chunked runner**；`agent-run` 只做试跑，其 Codex 会话预算
默认 1 小时足够，与全量耗时无关。

```bash
# 1) dataflowagent 交付试跑成功的 pipeline（不跑全量、不 merge）
loopai-obtainercli dm --root /path/to/warehouse dataflow agent-run \
  --target "score GSM8K answer-focused SFT rows and keep high-quality rows" \
  --dataset math_sft \
  --trial-rows 20 \
  --expected-outputs math_answer_quality \
  --recipe /path/to/recipe.yaml \
  --json

# 2) 上层 Codex 用交付的 pipeline 跑 chunk 全量（结果在 agent-run 的
#    upstream.chunked_run_command / apply_command 里）
python -m loopai.agents.Obtainer.datamixer.dataflow_chunked_runner \
  --input /path/to/full_input.jsonl \
  --pipeline /path/to/pipeline.py \
  --output /path/to/full_processed.jsonl --chunk-size 10000

# 3) 全量完成后合并回湖
loopai-obtainercli dm --root /path/to/warehouse apply-jsonl \
  --file /path/to/full_processed.jsonl --field content --json
```

DataFlowAgent agent-run rules:

- **Trial -> deliver -> upstream full is the contract.** The agent must
  trial-run the pipeline and deliver it (`mode=trial_run`, `pipeline_path` +
  `processed_jsonl`); it must NOT launch the full processing or write
  `full_processed.jsonl` itself. The upper-layer Codex runs the delivered
  pipeline over the exported full input and only treats L4 as complete when
  `full_processed.jsonl` exists and is verified.
- **Export the 1.5x bucket buffer, not the whole lake.** Pass `--recipe`
  (recipe.yaml) or `--mix-plan` (mix_plan.json) so the full input is sampled
  per bucket to `ceil(bucket_target * 1.5)` rows (fixed seed, short buckets
  export everything available). The processing scope is exactly
  `full_input.jsonl`; never re-export or widen it.
- **LLM quality-evaluation operators are mandatory.** Use DataFlow LLM
  scoring/filter operators (`PromptedEvaluator`, `PromptedFilter`, ...) for
  quality scoring. Cost/latency is NOT a valid reason to fall back to pure
  heuristic rules - a slow LLM pass just takes longer. Rule operators are
  allowed only when the task has no LLM-scoring semantics or the LLM serving is
  unavailable; say so in the summary. Preserve original fields and values.
  When post-training content needs construction or rewriting, use generation
  operators to add derived fields, then score and filter the generated content.
- **Full run is streaming, chunked, and executed by the upper layer.** The
  outer Codex drives the full scale through
  `loopai.agents.Obtainer.datamixer.dataflow_chunked_runner`
  (`--chunk-size 10000`, one chunk per pipeline launch, ordered merge) and must
  never load the whole export into a single DataFrame. The delivered pipeline
  must follow the `DATAFLOW_INPUT` / `DATAFLOW_CACHE_DIR` / `DATAFLOW_PREFIX`
  env-var convention so the scaffold can run it per chunk.
- **Never wrap agent-run or the chunked full run in a shell `timeout`**
  (e.g. `timeout 60 ...`). A shell timeout kills the inner Codex session or the
  chunked runner mid-flight and leaves the lake in a half-processed state. The
  **1-hour budget applies only to the agent-run Codex session (trial delivery)**;
  the upper-layer full run has no time budget and may take many hours when LLM
  quality-evaluation operators score every row - let it finish.
- The agent runs with its own Codex home (`codex_home_dataflow/AGENTS.md`),
  whose rules require it to deliver the trial-verified pipeline (never launch
  the full run itself) and to gate export on the 1.5x L4 redundancy floor
  (skipped when the user explicitly specifies an L3 export).

Index and recall:

```bash
loopai-obtainercli dm --root /path/to/warehouse index build --json
loopai-obtainercli dm --root /path/to/warehouse recall \
  --query "buggy and fixed Python code pairs for runtime exception repair" \
  --filter "domain = 'code' AND task_type = 'SFT'" \
  --limit 50 \
  --json
```

Lineage and snapshots:

```bash
loopai-obtainercli dm --root /path/to/warehouse snapshot create --name sft_mix_v1 --json
loopai-obtainercli dm --root /path/to/warehouse lineage list --json
```

## Internal Discovery Bridges

This low-level discovery bridge is for the isolated acquisition worker and for
human debugging only. If you are the outer Codex agent responding to a user
workflow request, skip this section and start `dataset-acquisition-agent`
instead. Do not create task JSON or run this command from the outer Codex
context.

```bash
loopai-obtainercli dm --root /path/to/warehouse dataset-acquisition-agent start \
  --run ./outputs/acquisition_run \
  --objective "collect buggy and fixed Python code-pair datasets covering syntax, logic, runtime, and assertion failures for SFT training" \
  --keywords "program repair dataset, buggy fixed code pairs, Python SyntaxError fix, runtime exception repair" \
  --target-datasets 8 \
  --max-rows-per-dataset 100000 \
  --max-bytes-per-dataset 2147483648 \
  --discovery-mode auto \
  --json
```

For multi-domain requests such as text2sql + math + code, describe the domain
split in `--objective` / `--keywords` / `--message`; the worker policy will
create isolated SearchAgent tasks and a WebAgent campaign internally, then run
the two discovery streams concurrently. 尽量使用镜像源；当 Hugging Face/Kaggle 等主站访问慢或不稳定时，
优先选择可用镜像或缓存源，并在 manifest/report 里记录实际来源。

## Manifest Download

This is the low-level download bridge used by the acquisition worker. Outer
Codex must not call `download manifest` during a normal workflow. Let
`dataset-acquisition-agent` materialize SearchAgent candidates into local
lake-ready files. It is not a lake operation.

Before downloading, compare the manifest against the original user request and
write a pruned manifest, for example `searchagent_manifest.filtered.json`.
Remove clearly unrelated candidates and keep a rejection report such as
`searchagent_manifest.rejections.json` with dataset id, reason, and the mismatch
dimension. Examples of rejection reasons: wrong domain, wrong task type, wrong
language, unrelated source family, missing target label shape, license blocker,
or provider failure risk.

For human debugging only, use `loopai-obtainercli download manifest ...` after
writing a filtered manifest and rejection report.

The downloader enforces a 100,000-row cap and a 2GiB local JSONL output cap per
dataset. `--max-rows 0` is also capped to 100,000 rows per dataset for safety.
When the byte cap is reached, the partial JSONL remains usable and the download
result must report the truncation. Production SFT sizing and final mixing must
be handled later through DataMixer recipes.

## Production SFT Export

For production SFT outflow, outer Codex should use the managed export worker
wrapper instead of manually driving `recipe validate/plan/preview/export`.
The wrapper starts an isolated Codex SDK worker and injects the detailed
DataMixer recipe, schema, validation, snapshot, and failure-handling policy into
that worker's context.

For heterogeneous SFT exports, schema mapping must be dataset/bucket-aware.
Do not use one global `output.sources` fallback order across datasets whose
fields have different semantics. Prefer bucket-level schema blocks such as
`recipe.buckets[].schema.fields` or `recipe.buckets[].export.schema.fields`.
Fields may be composed with templates when the final training row needs several
source fields, for example `output.template: "<think>{chain}</think>{answer}"`
for reasoning + answer, or for text2sql:
`instruction.template: "{question}"` and
`input.template: "{evidence}\n{sql_schema}\n{sql_block}"`.

Start a new isolated worker:

```bash
loopai-obtainercli dm --root /path/to/warehouse sft-export-agent start \
  --run ./outputs/sft_export_run \
  --analysis-report ./outputs/analyzer_report.md \
  --format alpaca \
  --target-records 100000 \
  --out ./outputs/sft_export_run/export
```

`start` returns after launching a background worker by default. Start the export
worker only after the DataFlowAgent post-processing stage has completed and the
final L4 dataset scale meets the recipe target; the lake must hold at least 1.5x
the target volume per bucket and overall before export is allowed. WebAgent and
the acquisition worker may still be active and continue adding data; their
terminal states are not export prerequisites. Use `--foreground` only when the
caller intentionally wants to block until the inner Codex SDK worker finishes.

Check a worker:

```bash
loopai-obtainercli dm --root /path/to/warehouse sft-export-agent status \
  --run ./outputs/sft_export_run
```

Continue the same inner Codex thread when the final report exposes a repairable
schema or quality problem:

```bash
loopai-obtainercli dm --root /path/to/warehouse sft-export-agent resume \
  --run ./outputs/sft_export_run \
  --message "Exclude buckets whose output field falls back to text, then re-export."
```

`resume` also runs in the background by default and returns a PID plus log
paths. Poll with `status`.

Do not pass `--model` to `sft-export-agent` unless the user explicitly requests
a one-off override. The worker should use Starter's configured Codex model by
default.

Outer Codex decides between `resume` and a fresh `start`:

- Use `resume` when the same worker understood the target but needs a bounded
  correction to recipe mapping, bucket filters, normalization, or validation.
- Use a fresh `start` when the worker context is polluted, picked the wrong
  task, or needs a different high-level strategy.

The worker wrapper owns the detailed constraints. In particular, for Alpaca SFT
it requires final rows to contain exactly `instruction`, `input`, and `output`,
forbids `output` fallback to whole-record text fields, rejects
`instruction == output`, requires DataMixer recipe export with snapshot, and
writes `final_report.json` with manifest, snapshot, digest, planned-versus-actual
bucket mix, validation evidence, and blockers. For datasets where a field like `output` is a noisy trace and
`answer` is the gold label, the worker must define that bucket's schema
explicitly instead of letting a global mapping choose the wrong source.

## End-To-End Agent Workflow

1. Read the Analyzer report or user request and extract the dataset intent.
2. Start `dataset-acquisition-agent`; it concurrently runs SearchAgent for
   hosted-dataset discovery and detached WebAgent for raw webpage L1 collection.
   WebAgent's L1 -> L2 -> L3 queues run continuously while candidate pruning,
   download, normalization, and DataMixer ingest proceed independently.
3. While both producers continue, poll current per-bucket record/token counts
   and quality gates. Treat lake sufficiency, not WebAgent or worker completion,
   as the transition condition for every downstream step.
4. As soon as those gates pass, run the mandatory DataFlowAgent stage
   (`dm dataflow agent-run`) for quality, deduplication, safety, and
   post-training validity; it delivers a trial-verified pipeline, then the
   outer Codex runs it over `full_input.jsonl` with the chunked runner
   (`dataflow_chunked_runner --chunk-size 10000`) and merges the L4 output
   with `apply-jsonl`. L4 must be produced before any export (unless the user
   explicitly requests an L3 export).
5. Build indexes when semantic recall or semantic deduplication is needed.
6. Start `sft-export-agent` for production recipe planning and export only after
   the DataFlowAgent stage completed and the L4 dataset scale meets the recipe
   target with at least 1.5x in-lake redundancy per bucket. Do not wait for
   WebAgent or `dataset-acquisition-agent` to reach a terminal state.
7. Poll `sft-export-agent status`; resume or restart based on blockers.
8. Poll `dataset-acquisition-agent status` independently; resume or restart it
   based on `final_report.json` and blockers without stopping downstream work.
9. Report warehouse path, datasets, record counts, processing results, recipe
    fingerprint, snapshot id, export path, and manifest path.

## Failure Handling

- Missing warehouse: run DataMixer `init` at the intended `--root`.
- Missing or unreliable semantic tags: do not export the requested taxonomy mix;
  tag/process more data first.
- Insufficient bucket size: report the exact bucket, available count/tokens, and
  target count/tokens from `recipe plan`.
- Download failure or empty selected file: stop before ingest.
- Unknown license or source: tag as unknown and avoid restricted training export
  unless explicitly approved.
- Embedding/index failure: report the failed DataMixer command and continue only
  if the requested recipe does not depend on semantic recall/deduplication.

## References

Detailed CLI usage:

```text
docs/OBTAINERCLI_USAGE.md
```