records#

Durable candidate lineage and observations for configurable post-MIP flows.

Classes

ArchitectureCandidate

ArtifactKind

CandidateLedger

Single-writer transactional registry backed by immutable node result files.

CandidateRevision

CandidateSet

NodeObservation

class ArchitectureCandidate#

Bases: object

__init__(architecture_id, block_configs, mip_metrics=<factory>, origins=<factory>, origin_revision_id='')#
Parameters:

  • architecture_id (str)

  • block_configs (list[Any])

  • mip_metrics (dict[str, float])

  • origins (list[dict[str, Any]])

  • origin_revision_id (str)

Return type:

None

architecture_id: str#
block_configs: list[Any]#
mip_metrics: dict[str, float]#
origin_revision_id: str = ''#
origins: list[dict[str, Any]]#
class ArtifactKind#

Bases: str, Enum

CHECKPOINT = 'checkpoint'#
CONFIG = 'config'#
__new__(value)#
class CandidateLedger#

Bases: object

Single-writer transactional registry backed by immutable node result files.

VERSION = 2#
__init__(root)#
Parameters:

root (str | Path)

active_profile_ids: set[str]#
add_revision(*, architecture_id, artifact_kind, artifact, parent_revision_id, producer_node)#
Parameters:

  • architecture_id (str)

  • artifact_kind (ArtifactKind)

  • artifact (Mapping[str, Any])

  • parent_revision_id (str | None)

  • producer_node (str)

Return type:

CandidateRevision

architectures: dict[str, ArchitectureCandidate]#
candidate_metadata(revision_id)#

Return architecture, revision lineage, and accumulated observations.

Parameters:

revision_id (str)

Return type:

dict[str, Any]

execution_identities: dict[str, str]#
ingest_mip(puzzle_dir)#
Parameters:

puzzle_dir (str | Path)

Return type:

None

load_candidate_set(node_id, *, execution_identity=None)#
Parameters:

  • node_id (str)

  • execution_identity (str | None)

Return type:

CandidateSet

load_execution_observations(execution_identities)#

Load node observations from exact immutable executions.

Parameters:

execution_identities (Mapping[str, str])

Return type:

None

observations: dict[str, dict[str, NodeObservation]]#
publish()#
Return type:

Path

publish_node(node_id, observations, candidate_set, execution_identity)#
Parameters:
Return type:

tuple[Path, Path]

resolve_concurrency_metrics(revision_id, reference)#

Resolve finite concurrency_<N>.<metric> values from one node observation.

Parameters:

  • revision_id (str)

  • reference (str)

Return type:

dict[int, float]

resolve_metric(revision_id, reference)#
Parameters:

  • revision_id (str)

  • reference (str)

Return type:

float | None

revisions: dict[str, CandidateRevision]#
root_set(flow_id, source)#
Parameters:

  • flow_id (str)

  • source (Mapping[str, Any])

Return type:

CandidateSet

source_revision(input_revision_id, source)#
Parameters:

  • input_revision_id (str)

  • source (str)

Return type:

CandidateRevision

class CandidateRevision#

Bases: object

__init__(revision_id, architecture_id, artifact_kind, artifact, parent_revision_id=None, producer_node='mip')#
Parameters:

  • revision_id (str)

  • architecture_id (str)

  • artifact_kind (ArtifactKind)

  • artifact (dict[str, Any])

  • parent_revision_id (str | None)

  • producer_node (str)

Return type:

None

architecture_id: str#
artifact: dict[str, Any]#
artifact_kind: ArtifactKind#
parent_revision_id: str | None = None#
producer_node: str = 'mip'#
revision_id: str#
class CandidateSet#

Bases: object

__init__(flow_id, node_id, revision_ids, identity, producer_execution_identity)#
Parameters:

  • flow_id (str)

  • node_id (str)

  • revision_ids (tuple[str, ...])

  • identity (str)

  • producer_execution_identity (str)

Return type:

None

classmethod create(flow_id, node_id, revision_ids, *, producer_execution_identity)#
Parameters:

  • flow_id (str)

  • node_id (str)

  • revision_ids (Iterable[str])

  • producer_execution_identity (str)

Return type:

CandidateSet

flow_id: str#
identity: str#
node_id: str#
producer_execution_identity: str#
revision_ids: tuple[str, ...]#
class NodeObservation#

Bases: object

__init__(node_id, input_revision_id, source_revision_id, output_revision_id, status, metrics=<factory>, artifacts=<factory>, warnings=<factory>, error=None)#
Parameters:

  • node_id (str)

  • input_revision_id (str)

  • source_revision_id (str)

  • output_revision_id (str | None)

  • status (str)

  • metrics (dict[str, float])

  • artifacts (dict[str, Any])

  • warnings (list[str])

  • error (str | None)

Return type:

None

artifacts: dict[str, Any]#
error: str | None = None#
input_revision_id: str#
metrics: dict[str, float]#
node_id: str#
output_revision_id: str | None#
source_revision_id: str#
status: str#
warnings: list[str]#