Public contracts

contracts

AmbiguousIdentityError

Bases: PublicSchemaValidationError

Raised when records cannot be associated with one unambiguous identity.

Source code in src/cofolder/modules/contracts/models.py

class AmbiguousIdentityError(PublicSchemaValidationError):
    """Raised when records cannot be associated with one unambiguous identity."""

EvidenceCompatibilityError

Bases: PublicSchemaValidationError

Raised when a computed metric uses an incompatible evidence regime.

Source code in src/cofolder/modules/contracts/models.py

class EvidenceCompatibilityError(PublicSchemaValidationError):
    """Raised when a computed metric uses an incompatible evidence regime."""

MetricRecordValidationError

Bases: PublicSchemaValidationError

Raised when a metric value or state contradicts its catalog definition.

Source code in src/cofolder/modules/contracts/models.py

class MetricRecordValidationError(PublicSchemaValidationError):
    """Raised when a metric value or state contradicts its catalog definition."""

PublicContractError

Bases: ValueError

Base class for invalid public-contract data.

Source code in src/cofolder/modules/contracts/models.py

class PublicContractError(ValueError):
    """Base class for invalid public-contract data."""

PublicSchemaValidationError

Bases: PublicContractError

Raised when a public record or bundle violates the schema.

Source code in src/cofolder/modules/contracts/models.py

class PublicSchemaValidationError(PublicContractError):
    """Raised when a public record or bundle violates the schema."""

PublicSerializationError

Bases: RuntimeError

Raised when a validated public bundle cannot be serialized atomically.

Source code in src/cofolder/modules/contracts/models.py

class PublicSerializationError(RuntimeError):
    """Raised when a validated public bundle cannot be serialized atomically."""

RunnerProvenanceError

Bases: ValueError

Raised when execution provenance is contradictory or malformed.

Source code in src/cofolder/modules/contracts/models.py

class RunnerProvenanceError(ValueError):
    """Raised when execution provenance is contradictory or malformed."""

ScreenExecutionCardinalityError

Bases: ScreenOutputNormalizationError

Raised when observed Screen output contradicts its execution matrix.

Source code in src/cofolder/modules/contracts/models.py

class ScreenExecutionCardinalityError(ScreenOutputNormalizationError):
    """Raised when observed Screen output contradicts its execution matrix."""

ScreenOutputNormalizationError

Bases: PublicContractError

Raised when child output cannot be mapped to planned Screen executions.

Source code in src/cofolder/modules/contracts/models.py

class ScreenOutputNormalizationError(PublicContractError):
    """Raised when child output cannot be mapped to planned Screen executions."""

SeedResolutionError

Bases: ValueError

Raised when seed resolution or backend adjustment is invalid.

Source code in src/cofolder/modules/contracts/models.py

class SeedResolutionError(ValueError):
    """Raised when seed resolution or backend adjustment is invalid."""

UnregisteredMetricError

Bases: PublicSchemaValidationError

Raised when a metric is absent from the public catalog.

Source code in src/cofolder/modules/contracts/models.py

class UnregisteredMetricError(PublicSchemaValidationError):
    """Raised when a metric is absent from the public catalog."""

WorkflowExecutionError

Bases: RuntimeError

Raised after a workflow has persisted a terminal failure bundle.

Source code in src/cofolder/modules/contracts/models.py

class WorkflowExecutionError(RuntimeError):
    """Raised after a workflow has persisted a terminal failure bundle."""

    def __init__(
        self,
        message: str,
        *,
        failures: tuple[WorkflowFailureRecord, ...],
        output_dir: Path,
    ) -> None:
        super().__init__(message)
        self.failures = failures
        self.output_dir = output_dir

read_public_records

read_public_records(path: str | Path) -> tuple[PublicRecord, ...]

Read typed public records from a records.jsonl file.

Source code in src/cofolder/modules/contracts/serialization.py

def read_public_records(path: str | Path) -> tuple[PublicRecord, ...]:
    """Read typed public records from a records.jsonl file."""
    records: list[PublicRecord] = []
    with Path(path).open(encoding="utf-8") as handle:
        for line in handle:
            if line.strip():
                records.append(record_from_dict(json.loads(line)))
    return tuple(records)

record_from_dict

record_from_dict(raw: dict[str, Any]) -> PublicRecord

Deserialize and validate one flattened public JSONL record.

Source code in src/cofolder/modules/contracts/serialization.py

def record_from_dict(raw: dict[str, Any]) -> PublicRecord:
    """Deserialize and validate one flattened public JSONL record."""
    identity_fields = OutputIdentity.__dataclass_fields__
    identity_values = {name: raw.get(name) for name in identity_fields}
    identity_values["workflow"] = WorkflowKind(identity_values["workflow"])
    if identity_values.get("backend_version_status") is not None:
        identity_values["backend_version_status"] = BackendVersionStatus(
            identity_values["backend_version_status"]
        )
    identity = OutputIdentity(**identity_values)
    kind = RecordKind(raw["record_kind"])
    envelope = PublicRecordEnvelope(
        schema_version=str(raw["schema_version"]),
        record_id=str(raw["record_id"]),
        record_kind=kind,
        identity=identity,
        created_at=str(raw["created_at"]),
    )
    artifacts = tuple(
        ArtifactReference(**item) for item in raw.get("artifacts") or ()
    )
    if kind is RecordKind.EXECUTION:
        error_value = raw.get("error")
        error = None
        if error_value is not None:
            error = StructuredExecutionError(
                stage=FailureStage(error_value["stage"]),
                exception_type=str(error_value["exception_type"]),
                error_code=str(error_value["error_code"]),
                message=str(error_value["message"]),
                retryable=bool(error_value.get("retryable", False)),
                details=error_value.get("details") or {},
            )
        record: PublicRecord = ExecutionRecord(
            envelope=envelope,
            status=ExecutionStatus(raw["status"]),
            execution_directory=str(raw["execution_directory"]),
            error=error,
            artifacts=artifacts,
        )
    elif kind is RecordKind.SUCCESS:
        record = SuccessRecord(envelope=envelope, artifacts=artifacts)
    elif kind is RecordKind.METRIC:
        record = MetricRecord(
            envelope=envelope,
            metric_name=str(raw["metric_name"]),
            metric_group=str(raw["metric_group"]),
            metric_class=MetricClass(raw["metric_class"]),
            status=RecordStatus(raw["status"]),
            value=raw.get("value"),
            unit=raw.get("unit"),
            direction=OptimizationDirection(raw["direction"]),
            evidence_regime=EvidenceRegime(raw["evidence_regime"]),
            evidence=tuple(
                EvidenceSource(**item) for item in raw.get("evidence") or ()
            ),
            statistic=raw.get("statistic", "value"),
            reason_code=raw.get("reason_code"),
            message=raw.get("message"),
        )
    elif kind is RecordKind.FAILURE:
        record = WorkflowFailureRecord(
            envelope=envelope,
            stage=FailureStage(raw["stage"]),
            exception_type=str(raw["exception_type"]),
            error_code=str(raw["error_code"]),
            message=str(raw["message"]),
            retryable=bool(raw.get("retryable", False)),
            details=raw.get("details") or {},
        )
    else:  # pragma: no cover - exhaustive over RecordKind
        raise ValueError(f"Unsupported public record kind: {kind!r}.")
    validate_public_record(record)
    return record