Batch experiments

kd.harness runs a matrix of fits and keeps what it ran. The loop is four calls: an ExperimentPlan of PlanEntry rows states the matrix, run_plan executes it into a fresh directory, EvidenceStore.load reopens that directory and re-verifies it, and build_consensus groups the runs by the law each one selected. merge_stores combines the stores of separate run_plan batches into one store before that last step. The store's guarantees and the meaning of the consensus axes are described under Batch experiments and evidence.

These names are imported from the subpackage rather than the package root: from kd.harness import ExperimentPlan, run_plan. Listed below are the entry points a caller writes; the module also carries the artifact machinery that run_plan and build_consensus drive on their own. The last six entries supervise a single worker command and its process group, for a caller that launches its own worker processes: ProcessSupervisor and run_supervised apply a time limit and a SIGTERM, grace, SIGKILL escalation to the whole group, and build_capped_command with check_memory_cap_available adds an optional cgroup memory cap.

Name	Summary
`kd.harness.ExperimentPlan(name, entries)`	An ordered execution matrix of plan entries with a content identity.
`kd.harness.PlanEntry(instrument, dataset_ref[, ...])`	One declarative episode: an instrument applied to a dataset at a seed.
`kd.harness.run_plan(plan, datasets[, ...])`	Pre-flight, then run every plan entry serially into a fresh store.
`kd.harness.PlanRunResult(store_root, outcomes)`	The outcome of a whole batch: the store root plus every episode outcome.
`kd.harness.EpisodeOutcome(entry_index, entry[, ...])`	The sealed result of running one plan entry.
`kd.harness.merge_stores(sources, root[, ...])`	Merge sealed stores into a new store at `root`; return it reloaded.
`kd.harness.EvidenceStore(root, plan[, ...])`	A fresh-dir, append-only, tamper-evident batch evidence directory.
`kd.harness.build_consensus(store, datasets[, ...])`	Aggregate a sealed evidence store into a consensus report tree.
`kd.harness.ConsensusReport(provenance, datasets[, ...])`	The full consensus report tree for one sealed evidence store.
`kd.harness.render_consensus_markdown(report, store)`	Render the report as a deterministic Markdown document (pure function).
`kd.harness.SKETCH_SIDECAR_FILENAME`	—
`kd.harness.ProcessSupervisor(command, env[, ...])`	Launch a command in its own session and supervise its whole process group.
`kd.harness.run_supervised(command, env[, ...])`	Launch a process group and block until cleanup or the post-SIGKILL bound.
`kd.harness.ProcessResult(pid, returncode[, ...])`	Observed leader termination, group cleanup, timing and sampled memory.
`kd.harness.ProcessEnding(value, names[, ...])`	Escalation stage reached while the leader was observed alive.
`kd.harness.build_capped_command(command, memory_max_gb)`	Prefix a command with a systemd user scope enforcing memory and no swap.
`kd.harness.check_memory_cap_available()`	Run a capped user scope to check whether memory caps are available.

ExperimentPlan `dataclass`

An ordered execution matrix of plan entries with a content identity.

from_dict `classmethod`

from_dict(data: dict[str, Any]) -> ExperimentPlan

Strictly decode a persisted plan payload.

plan_hash

plan_hash() -> str

Return the SHA-256 kd-plan-v1 identity of this plan.

to_dict

to_dict() -> dict[str, Any]

Return the pure content payload (no embedded hash).

PlanEntry `dataclass`

One declarative episode: an instrument applied to a dataset at a seed.

from_dict `classmethod`

from_dict(data: dict[str, Any]) -> PlanEntry

Strictly reconstruct an entry (validation re-runs in __init__).

to_dict

to_dict() -> dict[str, Any]

Return a defensive copy of the full JSON-native entry payload.

run_plan

run_plan(
    plan: ExperimentPlan,
    *,
    datasets: Mapping[str, PDEDataset | TabularDataset],
    store_root: Path,
    model_factory: Callable[..., Any] = Model,
    device: str | None = None,
    recording: RecordingOptions | None = None,
    catalog_path: Path | None = None,
) -> PlanRunResult

Pre-flight, then run every plan entry serially into a fresh store.

Parameters:

plan (ExperimentPlan) –

The ordered execution matrix to run.
datasets (Mapping[str, PDEDataset | TabularDataset]) –

Maps each dataset_ref to a loaded PDEDataset or a public TabularDataset (normalized once at entry through the same seam as Model.fit).
store_root (Path) –

Fresh directory for the EvidenceStore (must not exist or must be empty; append/resume is out of scope).
model_factory (Callable[..., Any], default: Model ) –

Facade constructor forwarded to run_episode; injectable so unit tests can stub the fit.
device (str | None, default: None ) –

Execution-environment parameter forwarded to run_episode and, from there, conditionally to the facade. None leaves the serial-execution call form literally unchanged.
recording (RecordingOptions | None, default: None ) –

When set, each entry runs inside a standard run dir <store_root>/runs/entry-NNNN/ (numbered by plan entry, matching records/entry-NNNN.json) wired per the options.
catalog_path (Path | None, default: None ) –

Cross-run ledger to append one kd-runcat-v1 row per outcome (requires recording; rows carry the entry index, the store's plan hash and paths relative to the catalog file).

Returns:

PlanRunResult –

A PlanRunResult with the store root and one outcome per entry.

Raises:

ValueError –

If pre-flight finds any unknown instrument or unresolved dataset_ref (all violations are collected into the message), or if catalog_path is given without recording.

PlanRunResult `dataclass`

The outcome of a whole batch: the store root plus every episode outcome.

EpisodeOutcome `dataclass`

The sealed result of running one plan entry.

merge_stores

merge_stores(
    sources: Sequence[Path | str],
    *,
    root: Path | str,
    name: str,
    version_note: str | None = None,
) -> EvidenceStore

Merge sealed stores into a new store at root; return it reloaded.

Parameters:

sources (Sequence[Path | str]) –

Roots of sealed stores, each written by one run_plan batch. Their plan entries are concatenated in this order.
root (Path | str) –

A fresh directory for the merged store (missing or empty).
name (str) –

The merged plan's name; it enters the kd-plan-v1 hash, so a single run_plan over the same entries under the same name has the same plan hash and entry indices.
version_note (str | None, default: None ) –

The merger's conclusion after reading the changelog between the sources' versions: what changed and whether it affects these results. Required when the sources differ in kd_version or git_sha; stored as given otherwise.

Returns:

EvidenceStore –

The merged store, loaded and fully verified (read-only). report.md is written into root.

EvidenceStore

A fresh-dir, append-only, tamper-evident batch evidence directory.

attempts `property`

attempts: tuple[dict[str, Any], ...]

Return the attempt ledger as a tuple of independent dict copies.

env `property`

env: dict[str, str]

Return a copy of the stored environment fingerprint.

plan `property`

plan: ExperimentPlan

Return the sealed experiment plan.

plan_hash `property`

plan_hash: str

Return the kd-plan-v1 identity of the sealed plan.

records `property`

records: MappingProxyType[int, RunRecord]

Return a read-only view of completed-episode records by entry index.

root `property`

root: Path

Return the store's root directory.

sources `property`

sources: tuple[dict[str, Any], ...]

Return the sources table of a merged store as independent row copies.

version_note `property`

version_note: str | None

Return the merger's note on differing kd versions (None if unset).

add_outcome

add_outcome(outcome: Any) -> None

Record one episode outcome and rewrite the index atomically.

create `classmethod`

create(
    root: Path,
    *,
    plan: ExperimentPlan,
    env: dict[str, str],
    sources: Sequence[Mapping[str, Any]] = (),
    version_note: str | None = None,
) -> EvidenceStore

Create a fresh store rooted at root.

load `classmethod`

load(root: Path) -> EvidenceStore

Load and fully verify a persisted store (returned READ-ONLY).

build_consensus

build_consensus(
    store: EvidenceStore,
    *,
    datasets: Mapping[str, PDEDataset] | None = None,
    policy: ConsensusPolicy = ConsensusPolicy(),
    instrument_schemas: Sequence[Mapping[str, Any]]
    | None = None,
    context_factory: VerifyContextFactory | None = None,
) -> ConsensusReport

Aggregate a sealed evidence store into a consensus report tree.

ConsensusReport `dataclass`

The full consensus report tree for one sealed evidence store.

render_consensus_markdown

render_consensus_markdown(
    report: ConsensusReport,
    *,
    store: EvidenceStore | None = None,
) -> str

Render the report as a deterministic Markdown document (pure function).

SKETCH_SIDECAR_FILENAME `module-attribute`

SKETCH_SIDECAR_FILENAME: Final[str] = 'sketch.json'

ProcessSupervisor

Launch a command in its own session and supervise its whole process group.

Parameters:

command (Sequence[str]) –

Executable and arguments, without shell expansion.
env (Mapping[str, str]) –

Complete environment for the command.
hard_seconds (float | None) –

Elapsed limit from launch, or None for no time limit.
grace_seconds (float) –

Wait after SIGTERM and again after SIGKILL, in seconds.
cwd (Path | None, default: None ) –

Working directory, or None to inherit the current directory.
stdout (IO[Any] | None, default: None ) –

Output handle, or None to inherit stdout. Owned by the caller.
stderr (IO[Any] | None, default: None ) –

Error handle, or None to inherit stderr. Owned by the caller.
memory_max_gb (float | None, default: None ) –

Optional cgroup memory cap in systemd's G units.

Raises:

OSError –

The command could not be launched.

pid `property`

pid: int

Leader PID, also the ID used to signal the entire group.

abort

abort() -> None

SIGKILL the group and wait at most one grace period for the leader.

hurry

hurry() -> None

End the SIGTERM grace now, making SIGKILL due on the next poll.

poll

poll() -> ProcessResult | None

Advance once, reaping the leader before probing group emptiness.

Returns:

ProcessResult | None –

None while supervision continues, otherwise its stable result.

Raises:

OSError –

An OS operation failed. A refused signal still marks its step; an interrupt during the send leaves it pending for retry.

request_stop

request_stop() -> None

Send SIGTERM immediately and let polling complete the escalation.

wait

wait(
    *,
    poll_seconds: float = 0.5,
    cancelled: Event | None = None,
) -> ProcessResult

Poll until completion; kill the group before propagating exceptions.

Parameters:

poll_seconds (float, default: 0.5 ) –

Interval between group probes and RSS samples.
cancelled (Event | None, default: None ) –

Optional shared event; setting it interrupts the wait.

Returns:

ProcessResult –

The final leader, group, timing and memory observations.

Raises:

CancelledError –

The cancellation event was set.

run_supervised

run_supervised(
    command: Sequence[str],
    *,
    env: Mapping[str, str],
    hard_seconds: float | None,
    grace_seconds: float,
    cwd: Path | None = None,
    stdout: IO[Any] | None = None,
    stderr: IO[Any] | None = None,
    memory_max_gb: float | None = None,
    poll_seconds: float = 0.5,
) -> ProcessResult

Launch a process group and block until cleanup or the post-SIGKILL bound.

Parameters:

command (Sequence[str]) –

Executable and arguments, without shell expansion.
env (Mapping[str, str]) –

Complete environment for the command.
hard_seconds (float | None) –

Elapsed limit from launch, or None for no time limit.
grace_seconds (float) –

Wait after SIGTERM and again after SIGKILL, in seconds.
cwd (Path | None, default: None ) –

Working directory, or None to inherit the current directory.
stdout (IO[Any] | None, default: None ) –

Output handle, or None to inherit stdout. Owned by the caller.
stderr (IO[Any] | None, default: None ) –

Error handle, or None to inherit stderr. Owned by the caller.
memory_max_gb (float | None, default: None ) –

Optional cgroup memory cap in systemd's G units.
poll_seconds (float, default: 0.5 ) –

Interval between group probes and RSS samples.

Returns:

ProcessResult –

The final leader, group, timing and memory observations.

Raises:

OSError –

Launch or a supervision operation failed.

ProcessResult `dataclass`

Observed leader termination, group cleanup, timing and sampled memory.

Attributes:

pid (int) –

Leader PID, also the supervised process group ID.
returncode (int | None) –

Leader exit status, or None if still alive at the bound.
ended_by (ProcessEnding) –

Last escalation stage reached while the leader was alive.
stop_reason (Literal['timeout', 'requested'] | None) –

timeout or requested when either initiated the escalation; None for natural exit and its descendant cleanup.
sigkill_used (bool) –

Whether the group SIGKILL step was taken, including descendant cleanup and an OS refusal of that signal.
started_at (str) –

Launch time as a UTC ISO 8601 string.
term_sent_at (str | None) –

UTC time of the SIGTERM step, or None.
kill_sent_at (str | None) –

UTC time of the SIGKILL step, or None.
ended_at (str) –

UTC time when supervision finished.
wallclock_seconds (float) –

Elapsed monotonic seconds from launch to completion.
peak_rss_bytes (int) –

Maximum sampled sum of group members' resident memory.
left_descendants (bool) –

Whether the leader exited before escalation and left members requiring cleanup.
group_emptied (bool) –

Whether the group was confirmed empty. False means the post-SIGKILL grace expired with members still present.

ProcessEnding

Bases: StrEnum

Escalation stage reached while the leader was observed alive.

build_capped_command

build_capped_command(
    command: Sequence[str], memory_max_gb: float | None
) -> list[str]

Prefix a command with a systemd user scope enforcing memory and no swap.

Parameters:

command (Sequence[str]) –

Executable and its arguments, passed without shell expansion.
memory_max_gb (float | None) –

MemoryMax in systemd's G units, or None for no wrapper.

Returns:

list[str] –

A new argument list, unchanged when no cap is requested.

check_memory_cap_available

check_memory_cap_available() -> None

Run a capped user scope to check whether memory caps are available.

Raises:

RuntimeError –

The scope failed; the message includes its stderr.
OSError –

The systemd-run executable could not be launched.

Batch experiments

ExperimentPlan dataclass

from_dict classmethod

plan_hash

to_dict

PlanEntry dataclass

from_dict classmethod

to_dict

run_plan

PlanRunResult dataclass

EpisodeOutcome dataclass

merge_stores

EvidenceStore

attempts property

env property

plan property

plan_hash property

records property

root property

sources property

version_note property

add_outcome

create classmethod

load classmethod

build_consensus

ConsensusReport dataclass

render_consensus_markdown

SKETCH_SIDECAR_FILENAME module-attribute

ProcessSupervisor

pid property

abort

hurry

poll

request_stop

wait

run_supervised

ProcessResult dataclass

ProcessEnding

build_capped_command

check_memory_cap_available

ExperimentPlan `dataclass`

from_dict `classmethod`

PlanEntry `dataclass`

from_dict `classmethod`

PlanRunResult `dataclass`

EpisodeOutcome `dataclass`

attempts `property`

env `property`

plan `property`

plan_hash `property`

records `property`

root `property`

sources `property`

version_note `property`

create `classmethod`

load `classmethod`

ConsensusReport `dataclass`

SKETCH_SIDECAR_FILENAME `module-attribute`

pid `property`

ProcessResult `dataclass`