Built-in `check-dataset` code

[ajtritt]

Part of #43. Depends on #45 (sample contract) and #44 (structured pipeline output).

A built-in code in src/dsagt/codes/check-dataset/ implementing the validation suite for a generated dataset. It is a code rather than agent instructions so that it runs under dsagt-run and its results land in trace_archive/ with provenance, consistent with the rest of DSAgt's audit trail.

Validation is what separates this feature from a prompt that writes a Dataset.

Checks

Check	Catches
Contract	ds[0] does not match declared keys, dtypes, shapes, ranges
Determinism	Same seed produces a different sample
Worker equivalence	num_workers=2 yields a different multiset than num_workers=0 (iterable duplication)
Split leakage	Group ids appearing in more than one split; declared sizes wrong
Throughput	Pathologically slow __getitem__
Model forward	Batch fails to pass through model.forward()
Contract staleness	The upstream pipeline changed after the contract was written

Scope

• Accepts an import path to the Dataset class, the contract file, and an output report path.
• Runs each check in a subprocess. The worker-equivalence check measures fork behavior, so importing the user's Dataset into the checking process would contaminate the state under observation (open file handles, an imported CUDA context, a seeded global RNG). Subprocess isolation also keeps a segfault in a native reader (HDF5, ADIOS2) from taking down the whole check run.
• Emits a JSON report to audit/ with per-check pass or fail plus measured values.
• Individual checks selectable, since the throughput check is slow.
• Staleness check: recompute the pipeline fingerprint from the current structured reconstruct_pipeline output and compare it against the value in the contract. A mismatch fails the check with a message naming which pipeline steps changed. Skipped in standalone mode, where the contract carries no fingerprint.

Note on the model-forward check: against the user's real model it verifies the consumer contract; against a generated reference model it only verifies that the contract is self-consistent and that tensors flow, since the reference model was derived from that same contract. It still catches dtype errors, ragged collation failures, and device mismatches, but a green result should not be over-read.

Automatic staleness detection on the heartbeat

The MCP server already runs a wall-clock heartbeat (_heartbeat in src/dsagt/mcp/server.py:115, 45 second interval) that ticks the trace collector and the code-use indexer in worker threads, logging failures without ever being fatal. Staleness detection becomes a third tick there.

Cost. Recomputing the fingerprint means loading and parsing every record in trace_archive/, which grows linearly with pipeline length, and most projects will never have a dataset contract at all. The tick is gated twice: it returns immediately if dataset_contract.yaml does not exist, and it recomputes only when the trace archive has changed since the last comparison (record count plus newest mtime). Steady state on a project with no contract is a single stat call every 45 seconds.

Delivery is the actual problem, not detection. The heartbeat is a background task inside the MCP server with no channel to the agent. Writing to the logger reaches nobody: the agent does not read the server's stderr, and MCP logging notifications are surfaced inconsistently across the five supported platforms, so a mechanism depending on them would work for some users and silently fail for others. Proposal, splitting detection from delivery:

• The heartbeat detects the mismatch and writes a marker to .dsagt/dataset_stale.json recording the old and new fingerprints and which steps changed.
• The marker is surfaced on the next tool response, the one channel guaranteed to work on every platform.
• check-dataset reads the same marker, so an explicit invocation and the automatic path report identically.

This is the part of the issue most likely to need discussion before implementation. Detection is straightforward; agreeing on how a background process interrupts an agent that is not asking is not.

Acceptance criteria

• Detects a deliberately introduced iterable-worker duplication bug.
• Detects a deliberately introduced group leak across splits.
• Detects an upstream pipeline change through the fingerprint mismatch.
• Emits a report parseable by the skill.