docs: add engine-native record selection plan

[nabinchha]

📋 Summary

Adds the source-of-truth design plan for engine-native record selection, allowing users to request an exact number of rows that satisfy a declared boolean criterion in one bounded, resumable DataDesigner run. The plan defines V1 as the complete user-facing feature and leaves concurrent batches and early cancellation as benchmark-driven optimizations.

🔗 Related Issue

Related to #790. The issue remains open to track implementation of the approved design.

🔄 Changes

• Add the engine-native record selection plan.
• Define the proposed RecordSelectionConfig API and exact accepted-row semantics.
• Separate logical candidate-batch progress from physical scheduler row groups.
• Specify durable checkpoints, resume behavior, exhaustion handling, metadata, processing order, and artifact cleanup.
• Document implementation phases, regression coverage motivated by feat: add repeat until workflow stages #773, risks, open questions, and the V1 definition of done.

🧪 Testing

• git diff --check origin/main..HEAD
• Markdown code and Mermaid fence balance validated
• Unit tests: N/A — design-plan-only change with no executable logic
• E2E tests: N/A — design-plan-only change with no executable logic

✅ Checklist

• Follows commit message conventions
• Commits are signed off (DCO)
• Architecture docs updated

[greptile-apps]

Greptile Summary

This PR adds the source-of-truth design plan for engine-native record selection, defining how DataDesigner.create() can reliably produce exactly X rows matching a declared boolean predicate within a bounded candidate budget. All three previously unresolved issues (media artifact strategy, IF_POSSIBLE fingerprint-mismatch state, and failed_generation_records accounting) were addressed in prior commits before this review.

• Config and API: Introduces RecordSelectionConfig with predicate_column, max_candidate_records, and on_exhausted; the explicit predicate-column model keeps DAG dependency discovery intact and avoids a duplicate expression runtime path.
• Artifact and resume model: Separates immutable selection-accepted/ partitions from the mutable parquet-files/ published view; uses atomic batch markers as the commit point; the post_generation_state machine (pending → started → complete) gates Hub publication independently of selection completion.
• Accounting invariant: Every committed marker satisfies candidate_records = accepted + rejected + null_predicate + failed_generation + trimmed_accepted, with trimmed_accepted_records and failed_generation_records explicitly tracked so the invariant holds even under partial generation failures and final-batch overshoot.

Confidence Score: 5/5

Documentation-only change adding a design plan with no executable code; safe to merge.

The plan is internally consistent: the candidate-accounting invariant holds in both the batch marker example and the global metadata example, the resume state machine covers all four transition paths (NEVER, no-checkpoints, match, mismatch×2 modes), and the trimmed_accepted_records and failed_generation_records fields are present in both the per-batch and aggregate metadata. The three issues raised in prior review threads were resolved before this submission. No executable logic is introduced.

No files require special attention; the single changed file is a design document.

Important Files Changed

Filename	Overview
plans/790/engine-native-record-selection.md	New 1,310-line design plan for engine-native record selection; logically consistent with previously addressed review concerns resolved in cafd13a and 1e992a5; no executable code changed.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant U as User
    participant I as DataDesigner.create
    participant AC as AcceptanceController
    participant S as AsyncTaskScheduler
    participant G as Column DAG
    participant P as ArtifactStorage

    U->>I: "create(builder, num_records=X)"
    I->>AC: "initialize(target=X, cap=M, resume_mode)"
    AC->>P: discover committed markers
    alt checkpoints exist and config/buffer/target match
        P-->>AC: reconstruct accepted_count, next_offset
    else no checkpoints
        AC->>AC: "Fresh start (offset=0)"
    else mismatch + IF_POSSIBLE
        AC->>P: clear selection-owned artifacts
        AC->>AC: "Fresh start (offset=0)"
    else mismatch + ALWAYS
        AC-->>I: IncompatibleResumeError
        I-->>U: error
    end

    loop "while accepted < X and candidates < cap"
        AC->>S: run CandidateBatch(offset, size)
        S->>G: generate column DAG
        G-->>S: candidate rows + predicate column
        S->>AC: select(dataframe)
        AC-->>S: SelectionDecision(accepted_indices, trimmed, ...)
        S->>P: promote accepted media (staging to committed)
        S->>P: write accepted partition (optional)
        S->>P: write batch marker (atomic commit)
        P-->>AC: durable accepted/candidate counts
    end

    alt "accepted == X"
        AC-->>I: exact dataset
        I-->>U: DatasetCreationResults
    else cap exhausted + return_partial
        AC-->>I: partial dataset
        I-->>U: DatasetCreationResults
    else cap exhausted + raise
        AC-->>I: RecordSelectionExhaustedError
        I-->>U: DataDesignerRecordSelectionExhaustedError
    end

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant U as User
    participant I as DataDesigner.create
    participant AC as AcceptanceController
    participant S as AsyncTaskScheduler
    participant G as Column DAG
    participant P as ArtifactStorage

    U->>I: "create(builder, num_records=X)"
    I->>AC: "initialize(target=X, cap=M, resume_mode)"
    AC->>P: discover committed markers
    alt checkpoints exist and config/buffer/target match
        P-->>AC: reconstruct accepted_count, next_offset
    else no checkpoints
        AC->>AC: "Fresh start (offset=0)"
    else mismatch + IF_POSSIBLE
        AC->>P: clear selection-owned artifacts
        AC->>AC: "Fresh start (offset=0)"
    else mismatch + ALWAYS
        AC-->>I: IncompatibleResumeError
        I-->>U: error
    end

    loop "while accepted < X and candidates < cap"
        AC->>S: run CandidateBatch(offset, size)
        S->>G: generate column DAG
        G-->>S: candidate rows + predicate column
        S->>AC: select(dataframe)
        AC-->>S: SelectionDecision(accepted_indices, trimmed, ...)
        S->>P: promote accepted media (staging to committed)
        S->>P: write accepted partition (optional)
        S->>P: write batch marker (atomic commit)
        P-->>AC: durable accepted/candidate counts
    end

    alt "accepted == X"
        AC-->>I: exact dataset
        I-->>U: DatasetCreationResults
    else cap exhausted + return_partial
        AC-->>I: partial dataset
        I-->>U: DatasetCreationResults
    else cap exhausted + raise
        AC-->>I: RecordSelectionExhaustedError
        I-->>U: DataDesignerRecordSelectionExhaustedError
    end

Loading

_{Reviews (9): Last reviewed commit: "Merge branch 'main' into codex/790-recor..." | Re-trigger Greptile}

[github-actions]

Code Review: PR #792 — docs: add engine-native record selection plan

Summary

This PR adds a single 985-line design document, plans/790/engine-native-record-selection.md,
proposing an engine-native record-selection feature: users declare a boolean predicate column and
request an exact number of accepted rows from one DataDesigner.create() call. The engine
generates bounded, immutable candidate batches, evaluates the predicate, checkpoints only accepted
rows, and supports durable resume.

Per the review instructions for plans/-only changes, this review focuses on plan completeness,
feasibility, and alignment with the existing architecture (verified against the codebase), not
linting or code style.

Overall this is a strong, unusually thorough design. It respects the interface → engine → config
dependency direction, correctly separates candidate coordinates from accepted-output coordinates,
and is grounded in real prior art (PR #773). The findings below are gaps and inconsistencies that
should be resolved before implementation, plus one significant reuse/altitude miss where the plan
proposes building infrastructure that already exists.

Findings

1. Proposed CandidateBatchPlan duplicates the existing ExplicitRowGroupPlan (reuse / altitude)

§2 (lines 397–410) proposes adding "a plan type capable of preserving explicit start offsets," a
new CandidateBatchPlan dataclass with __iter__ yielding (row_group_id, size) and a
row_group_start_offset(row_group) method. Phase 2 (line 812) lists row_group_plan.py — explicit candidate batch plan with start offset as new work.

This capability already exists. packages/.../dataset_builders/row_group_plan.py:244
defines ExplicitRowGroupPlan, which conforms to the RowGroupPlanLike protocol
(row_group_plan.py:11), materializes explicit _start_offsets per row group in __post_init__,
implements __iter__, and exposes row_group_start_offset() (line 289). The scheduler already
consumes this via _get_rg_start_offset() (async_scheduler.py:2148) and threads it through the
current_row_group_start_offset ContextVar (async_scheduler.py:1593). normalize_row_group_plan()
(line 300) is the existing entry point.

Since v1 maps each candidate batch to exactly one fresh row group of a known size at a known offset,
a candidate batch can be scheduled by constructing an ExplicitRowGroupPlan((row_group_id, size),)
with the desired base offset — no new plan type is needed. The plan should either reuse
ExplicitRowGroupPlan or justify why its interface is insufficient. As written it risks a parallel,
redundant implementation of already-tested offset infrastructure.

2. CandidateBatch and CandidateBatchPlan are near-identical DTOs (simplification)

The CandidateBatch dataclass (lines 343–348) and the proposed CandidateBatchPlan (lines 398–403)
carry the same four fields: candidate_batch_id, row_group_id, start_offset, size. Collapsing
these (and, per finding #1, folding scheduling onto ExplicitRowGroupPlan) removes a redundant type
and a source of drift between two structures that must stay in sync.

3. Batch-marker accepted_records vs trimmed_accepted_records semantics are undefined for resume

The batch marker (lines 567–579) records both accepted_records: 137 and
trimmed_accepted_records: 0, and resume "reconstruct[s] accepted and candidate progress from
committed batch markers" (line 625) by summing across markers. But the plan never states whether
accepted_records is the pre-trim count or the post-trim count actually written to parquet.

This matters on the overshoot path: the final candidate batch is trimmed to the remaining target
(lines 185, 373). If accepted_records is pre-trim and reconstruction sums accepted_records
across markers, the controller will compute a total larger than the rows actually persisted, and a
resume of a satisfied run could mis-detect the terminal state — or, worse, a crash between the
trimmed parquet write and marker commit could leave the reconstructed count inconsistent with the
parquet row count that "Verify every marker … points to a readable file with the expected accepted
count" (line 626) checks against. Define precisely: which field equals the parquet row count, and
which quantity the controller sums to decide accepted >= target.

4. Media-artifact cleanup is required for v1 but left undecided (completeness gap)

Line 694 states "V1 must choose and document one of these approaches" for rejected-row media
(orphaned image/audio/video/trace artifacts), and line 700 says "Do not silently accumulate unbounded
rejected media." Yet the decision is deferred to open question #5 (line 948), the "Recommended
decisions for v1" section (954–967) says nothing about media, and the Definition of Done (969–982)
omits it entirely. This is an internal contradiction: a concern the plan itself labels mandatory for
v1 has no committed decision and is absent from the completion criteria. Either commit to one
strategy (the plan already marks option 1, candidate-scoped staging, as "Preferred") and add it to
the DoD, or explicitly scope media-producing columns out of v1 selection.

5. buffer_size drives candidate batch sizing but is excluded from the fingerprint — resume interaction unaddressed

Candidate batch size derives from min(run_config.buffer_size, target, remaining_budget) (lines
422–428), and the plan places batch sizing in RunConfig deliberately (line 170). However
RunConfig/buffer_size is not part of DataDesignerConfig.fingerprint() — the fingerprint's
identity-relevant fields are columns, models, tools, seed, constraints, processors
(fingerprint.py:81–89), and only RecordSelectionConfig is added (line 169). Meanwhile the
existing resume contract already warns "buffer_size must match the original run"
(data_designer.py:232).

Consequence: a resume with a changed buffer_size will produce differently-sized future candidate
batches while past offsets are reconstructed from markers. Contiguity of candidate offsets is
preserved (each new batch starts where the last ended), so this may be acceptable — but the plan
never states it. Add an explicit note on whether buffer_size may change across a resumed selection
run and what the offset/sizing guarantee is. The resume test list (898–907) should cover it.

6. Phase decomposition omits the interface-layer num_records reinterpretation and run-boundary validation

The plan's central contract is that num_records changes meaning to "accepted output records" when
selection is configured (lines 21, 176). The create()-boundary validations
(max_candidate_records >= num_records, predicate existence, boolean output type — lines 190–197)
are correctly placed at the run boundary rather than in config (config cannot see the runtime
num_records). But the four implementation phases assign this nowhere: Phase 1 is config, Phase 2/3
are engine, Phase 4 is "Results, docs, and examples." The interface plumbing that routes
num_records as an accepted-target into DatasetBuilder.build(target_accepted=X) (per the sequence
diagram, line 283) and performs the run-boundary validation has no phase. Add an explicit
interface-layer deliverable so this isn't dropped.

7. Preview scope is left fuzzy relative to the Definition of Done

The Preview section (772–782) requires that preview(num_records=N) return accepted rows and honor
max_candidate_records, but offers two options (reuse controller in-memory vs. reject with a clear
error) without committing, and open question #4 restates the indecision. preview() today defaults
num_records=DEFAULT_NUM_RECORDS (data_designer.py:384), so an un-guarded preview under selection
would silently reinterpret that default as an accepted target and could run unbounded up to
max_candidate_records. The DoD (969–982) doesn't mention preview at all. Recommend committing to
the "reject clearly in v1" option in the DoD if preview integration isn't in scope, so the default
preview path can't become accidentally unbounded.

8. RecordSelectionExhaustedError is raised inside the engine builder loop but its home layer is an open question

The exhaustion example raises RecordSelectionExhaustedError(...) directly (lines 710–715), which by
the sequence diagram executes inside the engine DatasetBuilder — yet open question #8 (line 951)
still asks whether the error belongs "under interface errors or normalize[d as] an engine error at
the interface boundary." Per the project's "errors normalize at boundaries" invariant (AGENTS.md),
this should be resolved in the plan: define the error in the engine and re-expose/normalize it at the
interface, matching existing typed early-shutdown errors (which the plan references at line 727).
Leaving the layer undecided while already showing engine-side raise risks an implementation that
leaks an engine-internal type to callers.

Verdict

Approve with revisions requested (non-blocking for a design proposal).

This is a high-quality, implementable design that correctly respects the package layering and the
declarative-config contract. Before implementation begins, the author should resolve the reuse of
ExplicitRowGroupPlan (finding #1 — the most impactful, as it prevents building redundant offset
infrastructure), pin down the marker count semantics for resume (finding #3), and either commit to a
media-cleanup strategy or scope media out of v1 (finding #4). Findings #5–#8 are smaller completeness
gaps that would benefit from a sentence or a DoD/phase edit each. None of these block landing the
plan as a status: proposal document, but they should be addressed as the design moves toward
"approved."

[andreatnvidia]

The engine-native direction makes sense to me: exact accepted-row selection is a stage-local data-plane operation, while workflow chaining remains responsible for separate generate, judge, enrich, and transform stages.

The main design points I think need resolving before implementation are:

1. preserving explicit candidate offsets through the current row-group-plan normalization;
2. keeping selection checkpoint artifacts valid after after-generation processing; and
3. defining a successful zero-row return_partial path through the interface and profiler.

The remaining comments are API and lifecycle clarifications rather than objections to the overall direction.

[nabinchha]

Addressed the review feedback in cafd13af.

Key design revisions:

• Reuse and extend ExplicitRowGroupPlan(base_offset=...); remove the proposed duplicate CandidateBatchPlan.
• Define post-trim marker counts and a complete accounting invariant including failed-generation and trimmed rows.
• Separate immutable accepted partitions from mutable published output so after-generation re-chunking cannot invalidate resume.
• Commit to candidate/row-scoped media staging with accepted-only promotion and crash cleanup.
• Add explicit IF_POSSIBLE mismatch reset behavior and require matching buffer_size for resume.
• Add the missing interface phase, public error normalization, V1 preview rejection, and schema-bearing zero-row return_partial behavior.
• Restrict V1 predicates to row-local semantics, fix the nested judge-score example, and move log_pre_generation() outside the per-batch scheduler setup.

Validation: pre-commit hooks passed, git diff --check passed, and all Markdown/Mermaid fences are balanced. I also replied on each inline thread with its specific resolution.

[nabinchha]

@andreatgretel Thanks for the review. I addressed the three main lifecycle concerns and all of the inline clarifications:

• Candidate offsets now reuse ExplicitRowGroupPlan(base_offset=...), survive normalization, and have an ordered-seed regression test.
• Selection checkpoints reference immutable accepted partitions, while parquet-files/ is a separately materialized accepted-only published view that after-generation processing may re-chunk safely.
• Zero-row return_partial now has schema-bearing empty output, targeted zero-row guard bypass, and explicit no-profile behavior.
• V1 predicates are row-local, the judge example uses the full nested score path, and log_pre_generation() moves outside per-batch scheduler setup.

The latest follow-up (421c7e6f) also makes the accepted-only parquet-files/ contract explicit and specifies Hugging Face terminal-state validation, partial/empty dataset-card counts, media allowlisting, stale-shard cleanup, implementation ownership, and tests.

All pre-commit checks and Markdown fence validation pass. Re-requesting your review.

[greptile-apps]

resume=NEVER leaves stale batch markers that corrupt subsequent resumes

The state machine routes resume=NEVER directly to Fresh without cleaning up existing selection artifacts. If a prior run wrote markers 0–4 and a resume=NEVER run overwrites markers 0–2 but completes in only three batches, the old selection-checkpoints/batch_00003.json and batch_00004.json survive. A subsequent resume=ALWAYS or resume=IF_POSSIBLE that passes all fingerprint/buffer/target checks would enter Reconstruct, read all markers in selection-checkpoints/, and silently sum accepted counts from both runs—breaking the accepted-count invariant and potentially short-circuiting to Satisfied before the new run has actually reached its target.

The ResetForFresh state exists precisely to clear selection-owned artifacts before starting fresh (used by the IF_POSSIBLE mismatch path). The resume=NEVER transition should go through the same cleanup, or the plan should explicitly state that reconstruction is bounded by next_candidate_batch_id from global metadata rather than by the full contents of the selection-checkpoints/ directory.

Prompt To Fix With AI

This is a comment left during a code review.
Path: plans/790/engine-native-record-selection.md
Line: 780-786

Comment:
**`resume=NEVER` leaves stale batch markers that corrupt subsequent resumes**

The state machine routes `resume=NEVER` directly to `Fresh` without cleaning up existing selection artifacts. If a prior run wrote markers 0–4 and a `resume=NEVER` run overwrites markers 0–2 but completes in only three batches, the old `selection-checkpoints/batch_00003.json` and `batch_00004.json` survive. A subsequent `resume=ALWAYS` or `resume=IF_POSSIBLE` that passes all fingerprint/buffer/target checks would enter `Reconstruct`, read **all** markers in `selection-checkpoints/`, and silently sum accepted counts from both runs—breaking the accepted-count invariant and potentially short-circuiting to `Satisfied` before the new run has actually reached its target.

The `ResetForFresh` state exists precisely to clear selection-owned artifacts before starting fresh (used by the `IF_POSSIBLE` mismatch path). The `resume=NEVER` transition should go through the same cleanup, or the plan should explicitly state that reconstruction is bounded by `next_candidate_batch_id` from global metadata rather than by the full contents of the `selection-checkpoints/` directory.

How can I resolve this? If you propose a fix, please make it concise.

[nabinchha]

Good catch—the state diagram was ambiguous. ResumeMode.NEVER already follows ArtifactStorage fresh-run semantics: a non-empty requested dataset directory causes allocation of a new timestamped sibling, while an empty requested directory has no prior markers. Checkpoint discovery is scoped to that resolved artifact directory, so markers from separate runs cannot be combined. I updated the state machine and documented this invariant in 1e992a5. I also audited the implementation and reran the artifact-storage suite (69 tests passed), so no code change was required.