Skip to main content

​
synth_ai.sdk.optimization.job

Job lifecycle utilities for optimization jobs. This module provides utilities for tracking and emitting job lifecycle events using the OpenResponses-aligned event schema. It can be used by SDK clients to construct canonical events and track job state. Example: 
>>> from synth_ai.sdk.optimization.job import JobLifecycle, JobStatus
>>>
>>> # Create a job lifecycle tracker
>>> lifecycle = JobLifecycle(job_id="job_123")
>>>
>>> # Emit job started event
>>> event = lifecycle.start()
>>> print(event)  # {"type": "job.in_progress", "job_id": "job_123", ...}
>>>
>>> # Check status
>>> print(lifecycle.status)  # JobStatus.IN_PROGRESS
>>>
>>> # Emit completion event
>>> event = lifecycle.complete(data={"best_score": 0.95})

​
Classes

​
JobStatus

Job lifecycle status values. Methods:

​
from_string

from_string(cls, status: str) -> JobStatus

​
is_terminal

is_terminal(self) -> bool

​
is_success

is_success(self) -> bool

​
JobLifecycle

Track and emit job lifecycle events. This class provides a stateful wrapper around job lifecycle, making it easy to emit canonical events and track status transitions. It maintains an event history that can be used for debugging or persistence. Attributes:
  • job_id: The unique job identifier
  • status: Current job status
  • events: History of emitted events
  • started_at: Timestamp when job started (set on first start())
  • ended_at: Timestamp when job ended (set on complete/fail/cancel)
Methods:

​
start

start(self, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a job.in_progress event and update status. Args:
  • data: Optional event data (e.g., config info)
  • message: Optional human-readable message
Returns:
  • The job.in_progress event dictionary
Raises:
  • ValueError: If job is not in PENDING status

​
complete

complete(self, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a job.completed event and update status. Args:
  • data: Optional event data (e.g., results, best_score)
  • message: Optional human-readable message
Returns:
  • The job.completed event dictionary
Raises:
  • ValueError: If job is not in IN_PROGRESS status

​
fail

fail(self, error: Optional[str] = None, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a job.failed event and update status. Args:
  • error: Error message or description
  • data: Optional event data (e.g., traceback, error details)
  • message: Optional human-readable message
Returns:
  • The job.failed event dictionary
Raises:
  • ValueError: If job is not in IN_PROGRESS status

​
cancel

cancel(self, reason: Optional[str] = None, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a job.cancelled event and update status. Args:
  • reason: Cancellation reason
  • data: Optional event data
  • message: Optional human-readable message
Returns:
  • The job.cancelled event dictionary
Raises:
  • ValueError: If job is already in a terminal status

​
add_candidate

add_candidate(self, candidate_id: str, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a candidate.added event. Args:
  • candidate_id: Unique identifier for the candidate
  • data: Optional event data (e.g., candidate config)
  • message: Optional human-readable message
Returns:
  • The candidate.added event dictionary

​
complete_candidate

complete_candidate(self, candidate_id: str, score: Optional[float] = None, data: Optional[Dict[str, Any]] = None, message: Optional[str] = None) -> Dict[str, Any]
Emit a candidate.completed event. Args:
  • candidate_id: Unique identifier for the candidate
  • score: Optional evaluation score
  • data: Optional event data (e.g., metrics, evaluation results)
  • message: Optional human-readable message
Returns:
  • The candidate.completed event dictionary

​
is_terminal

is_terminal(self) -> bool
Check if the job is in a terminal status.

​
is_successful

is_successful(self) -> bool
Check if the job completed successfully.

​
elapsed_seconds

elapsed_seconds(self) -> Optional[float]
Get elapsed time in seconds (None if not started).