models

models

Core domain models for job execution summaries.

Contains Pydantic v2 models that provide the public contract between the execution engine (baton/runner) and the rest of the system.

JobCompletionSummary replaces the legacy RunSummary dataclass with a validated Pydantic model. The RunSummary alias in core.summary provides backward compatibility.

Classes

JobCompletionSummary

Bases: BaseModel

Summary of a completed job run.

Pydantic v2 model tracking key metrics for display at job completion: - Sheet success/failure/skip counts - Validation pass rate - Cost tracking - Duration and retry statistics - Hook execution results

Attributes

success_rate property

success_rate

Calculate sheet success rate as percentage.

Skipped sheets are excluded from the denominator since they were never attempted (e.g., skip_when_command conditions met).

Functions

from_checkpoint classmethod

from_checkpoint(checkpoint)

Construct a summary from checkpoint state.

Computes rates from sheet states, sums costs and durations.

Parameters:

Name	Type	Description	Default
checkpoint	CheckpointState	The checkpoint state to summarize.	required

Returns:

Type	Description
JobCompletionSummary	JobCompletionSummary with computed metrics.

Source code in src/marianne/core/models.py

@classmethod
def from_checkpoint(
    cls, checkpoint: CheckpointState
) -> JobCompletionSummary:
    """Construct a summary from checkpoint state.

    Computes rates from sheet states, sums costs and durations.

    Args:
        checkpoint: The checkpoint state to summarize.

    Returns:
        JobCompletionSummary with computed metrics.
    """
    completed = 0
    failed = 0
    skipped = 0
    total_cost = 0.0
    total_duration = 0.0
    validation_passed = 0
    validation_failed = 0
    successes_no_retry = 0

    for sheet_state in checkpoint.sheets.values():
        if sheet_state.status == SheetStatus.COMPLETED:
            completed += 1
            if sheet_state.success_without_retry:
                successes_no_retry += 1
        elif sheet_state.status == SheetStatus.FAILED:
            failed += 1
        elif sheet_state.status == SheetStatus.SKIPPED:
            skipped += 1

        if sheet_state.validation_passed is True:
            validation_passed += 1
        elif sheet_state.validation_passed is False:
            validation_failed += 1

        total_cost += sheet_state.total_cost_usd
        total_duration += sheet_state.total_duration_seconds

    total_validations = validation_passed + validation_failed
    val_rate = (
        (validation_passed / total_validations * 100)
        if total_validations > 0
        else 100.0
    )
    no_retry_rate = (
        (successes_no_retry / completed * 100) if completed > 0 else 0.0
    )

    return cls(
        job_id=checkpoint.job_id,
        job_name=checkpoint.job_name,
        total_sheets=checkpoint.total_sheets,
        completed_sheets=completed,
        failed_sheets=failed,
        skipped_sheets=skipped,
        total_cost_usd=total_cost,
        total_duration_seconds=total_duration,
        validation_pass_rate=val_rate,
        success_without_retry_rate=no_retry_rate,
        validation_pass_count=validation_passed,
        validation_fail_count=validation_failed,
        successes_without_retry=successes_no_retry,
        final_status=checkpoint.status,
        total_input_tokens=checkpoint.total_input_tokens,
        total_output_tokens=checkpoint.total_output_tokens,
        total_estimated_cost=checkpoint.total_estimated_cost,
        cost_limit_hit=checkpoint.cost_limit_reached,
    )

to_dict

to_dict()

Convert summary to dictionary for JSON output.

Source code in src/marianne/core/models.py

def to_dict(self) -> dict[str, Any]:
    """Convert summary to dictionary for JSON output."""
    return {
        "job_id": self.job_id,
        "job_name": self.job_name,
        "status": self.final_status.value,
        "duration_seconds": round(self.total_duration_seconds, 2),
        "duration_formatted": self._format_duration(
            self.total_duration_seconds
        ),
        "sheets": {
            "total": self.total_sheets,
            "completed": self.completed_sheets,
            "failed": self.failed_sheets,
            "skipped": self.skipped_sheets,
            "success_rate": round(self.success_rate, 1),
        },
        "validation": {
            "passed": self.validation_pass_count,
            "failed": self.validation_fail_count,
            "pass_rate": round(self.validation_pass_rate, 1),
        },
        "execution": {
            "total_retries": self.total_retries,
            "completion_attempts": self.total_completion_attempts,
            "rate_limit_waits": self.rate_limit_waits,
            "successes_without_retry": self.successes_without_retry,
            "success_without_retry_rate": round(
                self.success_without_retry_rate, 1
            ),
        },
    }