Artifact and Serialization API Reference

ArtifactIdentity

async_batch_llm.ArtifactIdentity dataclass

ArtifactIdentity(provider: str | None = None, model: str | None = None, prompt_version: str | None = None, parser_version: str | None = None, application_version: str | None = None, extra: Mapping[str, JSONValue] = dict())

Caller-supplied provenance used to decide whether replay is compatible.

ResumePolicy

async_batch_llm.ResumePolicy

Bases: str, Enum

Which compatible terminal artifact records may bypass provider work.

ArtifactStore

async_batch_llm.ArtifactStore

Bases: Protocol

Provider-neutral asynchronous checkpoint/replay store.

append async

append(work_item: LLMWorkItem[Any, Any, Any], prepared_item: Any, result: WorkItemResult[Any, Any]) -> None

Durably append one newly executed terminal result.

Source code in src/async_batch_llm/artifacts.py

async def append(
    self,
    work_item: LLMWorkItem[Any, Any, Any],
    prepared_item: Any,
    result: WorkItemResult[Any, Any],
) -> None:
    """Durably append one newly executed terminal result."""

close async

close() -> None

Flush and close the store; repeated calls are safe.

Source code in src/async_batch_llm/artifacts.py

async def close(self) -> None:
    """Flush and close the store; repeated calls are safe."""

iter_results

iter_results(*, successes_only: bool = False) -> AsyncIterator[WorkItemResult]

Iterate stored results without starting a processor.

Source code in src/async_batch_llm/artifacts.py

def iter_results(self, *, successes_only: bool = False) -> AsyncIterator[WorkItemResult]:
    """Iterate stored results without starting a processor."""

lookup async

lookup(work_item: LLMWorkItem[Any, Any, Any], prepared_item: Any, policy: ResumePolicy) -> WorkItemResult[Any, Any] | None

Return the newest compatible reusable result, if any.

Source code in src/async_batch_llm/artifacts.py

async def lookup(
    self,
    work_item: LLMWorkItem[Any, Any, Any],
    prepared_item: Any,
    policy: ResumePolicy,
) -> WorkItemResult[Any, Any] | None:
    """Return the newest compatible reusable result, if any."""

prepare_item async

prepare_item(work_item: LLMWorkItem[Any, Any, Any]) -> Any

Prepare the run and validate/fingerprint an item before execution.

Source code in src/async_batch_llm/artifacts.py

async def prepare_item(self, work_item: LLMWorkItem[Any, Any, Any]) -> Any:
    """Prepare the run and validate/fingerprint an item before execution."""

JsonlArtifactStore

async_batch_llm.JsonlArtifactStore

JsonlArtifactStore(path: str | Path, *, identity: ArtifactIdentity | None = None, user_metadata: Mapping[str, JSONValue] | None = None, include_output: bool = True, include_metadata: bool = True, include_prompt: bool = False, include_context: bool = False, context_in_identity: bool = True, encoder: ValueEncoder | None = None, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None, context_fingerprinter: ContextFingerprinter | None = None, cost_calculator: CostCalculator | None = None, fsync: bool = False)

Append-only version-1 JSONL artifact store.

Prompt and context text are excluded by default; their SHA-256 hashes are always recorded for compatibility. Output and metadata are included by default because successful replay requires output. Set include_output false for audit-only artifacts; those successful records are not replayable.

Source code in src/async_batch_llm/artifacts.py

def __init__(
    self,
    path: str | Path,
    *,
    identity: ArtifactIdentity | None = None,
    user_metadata: Mapping[str, JSONValue] | None = None,
    include_output: bool = True,
    include_metadata: bool = True,
    include_prompt: bool = False,
    include_context: bool = False,
    context_in_identity: bool = True,
    encoder: ValueEncoder | None = None,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
    context_fingerprinter: ContextFingerprinter | None = None,
    cost_calculator: CostCalculator | None = None,
    fsync: bool = False,
) -> None:
    self.path = Path(path)
    self.identity = identity
    self._identity_is_explicit = identity is not None
    self.user_metadata = user_metadata or {}
    self.include_output = include_output
    self.include_metadata = include_metadata
    self.include_prompt = include_prompt
    self.include_context = include_context
    self.context_in_identity = context_in_identity
    self.encoder = encoder
    self.output_decoder = output_decoder
    self.context_decoder = context_decoder
    self.context_fingerprinter = context_fingerprinter
    self.cost_calculator = cost_calculator
    self.fsync = fsync

    # With no explicit identity, resolution is deferred to the first
    # prepare_item() call, which infers provider/model from the item's
    # strategy. Until then the
    # fingerprint is unset and lookup/append refuse to run.
    self._identity_value: dict[str, JSONValue] | None = None
    self.identity_fingerprint: str | None = None
    if identity is not None:
        try:
            self._identity_value, self.identity_fingerprint = fingerprint_identity(identity)
        except ResultSerializationError as exc:
            raise ArtifactSerializationError(str(exc)) from exc
    try:
        metadata_value = to_json_value(self.user_metadata, path="$.user_metadata")
        _encode_canonical_utf8(metadata_value, path="$.user_metadata")
    except ResultSerializationError as exc:
        raise ArtifactSerializationError(str(exc)) from exc
    if not isinstance(metadata_value, dict):
        raise ArtifactSerializationError("user_metadata must serialize to an object")
    self._user_metadata_value = metadata_value

    self._lock = asyncio.Lock()
    self._prepared = False
    self._closed = False
    self._fatal_write_error: Exception | None = None
    self._handle: TextIO | None = None
    self._records: list[dict[str, Any]] = []
    self._latest_replayable: dict[_ReplayKey, dict[str, Any]] = {}
    self._latest_success: dict[_ReplayKey, dict[str, Any]] = {}
    self._next_sequence = 0
    self._detached_io_tasks: set[asyncio.Task[Any]] = set()
    self._detached_io_errors: list[Exception] = []

iter_results async

iter_results(*, successes_only: bool = False) -> AsyncIterator[WorkItemResult[Any, Any]]

Stream a finite, bounded-memory snapshot of complete item records.

Source code in src/async_batch_llm/artifacts.py

async def iter_results(
    self, *, successes_only: bool = False
) -> AsyncIterator[WorkItemResult[Any, Any]]:
    """Stream a finite, bounded-memory snapshot of complete item records."""
    if not self._prepared:
        try:
            exists_and_nonempty = await asyncio.to_thread(
                lambda: self.path.exists() and self.path.stat().st_size > 0
            )
        except OSError as exc:
            raise ArtifactIOError(f"Could not inspect artifact {self.path}: {exc}") from exc
        if not exists_and_nonempty:
            # Preserve the historical create-on-first-operation behavior
            # for an explicitly identified empty store.
            await self._prepare()

    async def capture_snapshot() -> int:
        async with self._lock:
            self._raise_detached_io_error()
            return await asyncio.to_thread(_artifact_snapshot_size, self.path)

    snapshot_size = await self._run_lock_owner(capture_snapshot())
    offset = 0
    line_number = 0
    manifest_seen = False
    done = False
    while not done:
        page, offset, line_number, manifest_seen, done = await asyncio.to_thread(
            _read_artifact_page,
            self.path,
            offset=offset,
            snapshot_size=snapshot_size,
            line_number=line_number,
            manifest_seen=manifest_seen,
            page_size=_JSONL_ITER_PAGE_SIZE,
        )
        for record in page:
            if successes_only and not record.get("success"):
                continue
            try:
                yield decode_stored_result(
                    record,
                    output_decoder=self.output_decoder,
                    context_decoder=self.context_decoder,
                )
            except (KeyError, ResultSerializationError) as exc:
                sequence = record.get("record_sequence")
                if isinstance(sequence, int) and not isinstance(sequence, bool):
                    sequence += 1
                item_id = record.get("item_id")
                raise ArtifactFormatError(
                    f"Malformed stored result at sequence {sequence!r} "
                    f"for item {item_id!r}: {exc}"
                ) from exc

prepare_item async

prepare_item(work_item: LLMWorkItem[Any, Any, Any]) -> _ItemFingerprint

Create/validate the artifact and fingerprint input before provider work.

Source code in src/async_batch_llm/artifacts.py

async def prepare_item(self, work_item: LLMWorkItem[Any, Any, Any]) -> _ItemFingerprint:
    """Create/validate the artifact and fingerprint input before provider work."""
    self._resolve_identity_from(work_item.strategy)
    await self._prepare()
    return self._fingerprint_item(work_item)

read_results classmethod

read_results(path: str | Path, *, successes_only: bool = False, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None) -> BatchResult[Any, Any]

Read stored results without opening a writer or calling a provider.

Source code in src/async_batch_llm/artifacts.py

@classmethod
def read_results(
    cls,
    path: str | Path,
    *,
    successes_only: bool = False,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
) -> BatchResult[Any, Any]:
    """Read stored results without opening a writer or calling a provider."""
    records = _read_artifact_records(Path(path), allow_create=False).records
    results: list[WorkItemResult[Any, Any]] = []
    for record in records:
        if successes_only and not record.get("success"):
            continue
        try:
            results.append(
                decode_stored_result(
                    record,
                    output_decoder=output_decoder,
                    context_decoder=context_decoder,
                )
            )
        except (KeyError, ResultSerializationError) as exc:
            sequence = record.get("record_sequence")
            if isinstance(sequence, int) and not isinstance(sequence, bool):
                sequence += 1
            item_id = record.get("item_id")
            raise ArtifactFormatError(
                f"Malformed stored result at sequence {sequence!r} for item {item_id!r}: {exc}"
            ) from exc
    return BatchResult(results=results, termination=BatchTermination())

SqliteArtifactStore

Indexed SQLite backend for large restartable runs (v0.21). Note that SqliteArtifactStore.read_results() is an async classmethod, unlike the synchronous JSONL convenience. It materializes a finite high-water snapshot through an operationally read-only path connection. Normal reads participate in SQLite locking and may create empty WAL/SHM coordination sidecars so a writer can safely start during inspection; live store.iter_results() instead flushes and inspects that writable store.

async_batch_llm.SqliteArtifactStore

SqliteArtifactStore(path: str | Path, *, identity: ArtifactIdentity | None = None, user_metadata: Mapping[str, JSONValue] | None = None, include_output: bool = True, include_metadata: bool = True, include_prompt: bool = False, include_context: bool = False, context_in_identity: bool = True, encoder: ValueEncoder | None = None, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None, context_fingerprinter: ContextFingerprinter | None = None, cost_calculator: CostCalculator | None = None, durability: SqliteDurability = SqliteDurability.BALANCED, commit_batch_size: int = 100, commit_interval_seconds: float = 0.01, busy_timeout_seconds: float = 5.0, read_batch_size: int = 1000)

Indexed, batched artifact store backed by standard-library SQLite.

One writable instance owns one connection, one executor thread, and one async writer task. It supports sequential reopen but is not a distributed work-claiming or exactly-once execution system.

Source code in src/async_batch_llm/sqlite_artifacts.py

def __init__(
    self,
    path: str | Path,
    *,
    identity: ArtifactIdentity | None = None,
    user_metadata: Mapping[str, JSONValue] | None = None,
    include_output: bool = True,
    include_metadata: bool = True,
    include_prompt: bool = False,
    include_context: bool = False,
    context_in_identity: bool = True,
    encoder: ValueEncoder | None = None,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
    context_fingerprinter: ContextFingerprinter | None = None,
    cost_calculator: CostCalculator | None = None,
    durability: SqliteDurability = SqliteDurability.BALANCED,
    commit_batch_size: int = 100,
    commit_interval_seconds: float = 0.01,
    busy_timeout_seconds: float = 5.0,
    read_batch_size: int = 1000,
) -> None:
    raw_path = os.fspath(path)
    if raw_path == ":memory:":
        raise ValueError(
            "SqliteArtifactStore requires a filesystem path; ':memory:' is unsupported"
        )
    if isinstance(raw_path, str) and raw_path.startswith("file:"):
        raise ValueError("SQLite URI filenames are unsupported; pass a normal filesystem path")
    self.path = Path(path)
    self.identity = identity
    self._identity_is_explicit = identity is not None
    self.user_metadata = user_metadata or {}
    self.include_output = include_output
    self.include_metadata = include_metadata
    self.include_prompt = include_prompt
    self.include_context = include_context
    self.context_in_identity = context_in_identity
    self.encoder = encoder
    self.output_decoder = output_decoder
    self.context_decoder = context_decoder
    self.context_fingerprinter = context_fingerprinter
    self.cost_calculator = cost_calculator
    try:
        self.durability = SqliteDurability(durability)
    except (TypeError, ValueError) as exc:
        choices = ", ".join(value.value for value in SqliteDurability)
        raise ValueError(f"durability must be one of: {choices} (got {durability!r})") from exc
    self.commit_batch_size = _validate_positive_integer("commit_batch_size", commit_batch_size)
    self.commit_interval_seconds = _validate_non_negative_number(
        "commit_interval_seconds", commit_interval_seconds
    )
    self.busy_timeout_seconds = _validate_non_negative_number(
        "busy_timeout_seconds", busy_timeout_seconds
    )
    self.read_batch_size = _validate_positive_integer("read_batch_size", read_batch_size)

    try:
        metadata_value = to_json_value(self.user_metadata, path="$.user_metadata")
        _encode_canonical_utf8(metadata_value, path="$.user_metadata")
    except ResultSerializationError as exc:
        raise ArtifactSerializationError(str(exc)) from exc
    if not isinstance(metadata_value, dict):
        raise ArtifactSerializationError("user_metadata must serialize to an object")
    self._user_metadata_value = metadata_value

    self._identity_value: dict[str, JSONValue] | None = None
    self.identity_fingerprint: str | None = None
    if identity is not None:
        try:
            self._identity_value, self.identity_fingerprint = fingerprint_identity(identity)
        except ResultSerializationError as exc:
            raise ArtifactSerializationError(str(exc)) from exc

    self._executor = ThreadPoolExecutor(
        max_workers=1,
        thread_name_prefix="async-batch-llm-sqlite",
    )
    self._connection: sqlite3.Connection | None = None
    self._prepare_task: asyncio.Task[None] | None = None
    self._prepare_error_observed = False
    self._identity_tasks: dict[str, asyncio.Task[None]] = {}
    self._writer_task: asyncio.Task[None] | None = None
    self._active_writer_batch: list[_AppendRequest] = []
    self._writer_queue: asyncio.Queue[_WriterRequest] = asyncio.Queue()
    self._close_task: asyncio.Task[None] | None = None
    self._closing = False
    self._closed = False
    self._fatal_error: ArtifactError | None = None
    self._fatal_error_observed = False
    self._detached_error: ArtifactError | None = None
    self._detached_futures: set[asyncio.Future[None]] = set()
    self._close_errors: list[ArtifactError] = []
    self._executor_shutdown = False
    self._wal_autocheckpoint_pages = _WAL_AUTOCHECKPOINT_PAGES
    self._effective_wal_autocheckpoint_pages: int | None = None
    self._page_size: int | None = None
    self._transaction_count = 0
    self._last_checkpoint_busy: bool | None = None

close async

close() -> None

Drain writes, checkpoint/truncate WAL, and terminate the executor.

Source code in src/async_batch_llm/sqlite_artifacts.py

async def close(self) -> None:
    """Drain writes, checkpoint/truncate WAL, and terminate the executor."""
    if self._close_task is None:
        self._closing = True
        self._close_task = asyncio.create_task(self._close_impl())
        self._close_task.add_done_callback(self._consume_task_exception)
    task = self._close_task
    try:
        await _await_without_cancelling(task)
    except asyncio.CancelledError:
        raise
    self._raise_next_close_error()

iter_results

iter_results(*, successes_only: bool = False) -> AsyncIterator[WorkItemResult[Any, Any]]

Return a bounded keyset iterator over a finite committed snapshot.

Source code in src/async_batch_llm/sqlite_artifacts.py

def iter_results(
    self, *, successes_only: bool = False
) -> AsyncIterator[WorkItemResult[Any, Any]]:
    """Return a bounded keyset iterator over a finite committed snapshot."""
    return self._iter_results(successes_only=successes_only)

lookup async

lookup(work_item: LLMWorkItem[Any, Any, Any], prepared_item: Any, policy: ResumePolicy) -> WorkItemResult[Any, Any] | None

Look up the newest compatible terminal row through a replay index.

Source code in src/async_batch_llm/sqlite_artifacts.py

async def lookup(
    self,
    work_item: LLMWorkItem[Any, Any, Any],
    prepared_item: Any,
    policy: ResumePolicy,
) -> WorkItemResult[Any, Any] | None:
    """Look up the newest compatible terminal row through a replay index."""
    if policy is ResumePolicy.NONE:
        return None
    await self._before_operation()
    prepared = self._coerce_prepared(prepared_item)
    await self._prepare()
    _, _, identity_fingerprint = self._require_resolved_identity()
    row = await self._run_db(
        self._lookup_sync,
        work_item.item_id,
        prepared,
        identity_fingerprint,
        policy,
    )
    if row is None or is_non_replayable_record(row):
        return None
    try:
        return restore_replayed_result(
            row,
            work_item,
            output_decoder=self.output_decoder,
            context_decoder=self.context_decoder,
        )
    except (KeyError, ResultSerializationError) as exc:
        raise ArtifactFormatError(
            f"Malformed stored result at sequence {row.get('record_sequence')!r} "
            f"for item {work_item.item_id!r}: {exc}"
        ) from exc

prepare_item async

prepare_item(work_item: LLMWorkItem[Any, Any, Any]) -> PreparedArtifactItem

Open/validate the database and fingerprint input before provider work.

Source code in src/async_batch_llm/sqlite_artifacts.py

async def prepare_item(self, work_item: LLMWorkItem[Any, Any, Any]) -> PreparedArtifactItem:
    """Open/validate the database and fingerprint input before provider work."""
    await self._before_operation()
    self._resolve_identity_from(work_item.strategy)
    await self._prepare()
    _, _, fingerprint = self._require_resolved_identity()
    registration = self._identity_tasks.get(fingerprint)
    if registration is None:
        registration = asyncio.create_task(self._register_identity())
        registration.add_done_callback(self._consume_task_exception)
        self._identity_tasks[fingerprint] = registration
    await self._await_owned_future(registration)
    try:
        return fingerprint_work_item(
            work_item,
            context_in_identity=self.context_in_identity,
            encoder=self.encoder,
            context_fingerprinter=self.context_fingerprinter,
        )
    except ResultSerializationError as exc:
        raise ArtifactSerializationError(str(exc)) from exc
    except Exception as exc:
        raise ArtifactSerializationError(
            f"Context fingerprinter failed for item {work_item.item_id!r}: {exc}"
        ) from exc

read_results async classmethod

read_results(path: str | Path, *, successes_only: bool = False, output_decoder: ValueDecoder | None = None, context_decoder: ValueDecoder | None = None, read_batch_size: int = 1000) -> BatchResult[Any, Any]

Materialize a finite snapshot through an operationally read-only connection.

Source code in src/async_batch_llm/sqlite_artifacts.py

@classmethod
async def read_results(
    cls,
    path: str | Path,
    *,
    successes_only: bool = False,
    output_decoder: ValueDecoder | None = None,
    context_decoder: ValueDecoder | None = None,
    read_batch_size: int = 1000,
) -> BatchResult[Any, Any]:
    """Materialize a finite snapshot through an operationally read-only connection."""
    artifact_path = Path(path)
    page_size = _validate_positive_integer("read_batch_size", read_batch_size)
    executor = ThreadPoolExecutor(
        max_workers=1,
        thread_name_prefix="async-batch-llm-sqlite-reader",
    )
    loop = asyncio.get_running_loop()
    future = loop.run_in_executor(
        executor,
        cls._read_results_sync,
        artifact_path,
        successes_only,
        output_decoder,
        context_decoder,
        page_size,
    )
    try:
        return cast(BatchResult[Any, Any], await _await_without_cancelling(future))
    except asyncio.CancelledError:
        # The SQLite connection belongs to the worker thread. Let its
        # operation reach the helper's finally block before propagating
        # cancellation when the caller allows cleanup to finish. A repeated
        # cancellation remains prompt: the executor is then told to shut
        # down without synchronously blocking the event loop.
        try:
            await _await_without_cancelling(future)
        except asyncio.CancelledError:
            pass
        except Exception:
            # Cancellation remains the public outcome, but retrieving the
            # worker exception prevents an unobserved background failure.
            pass
        raise
    finally:
        if not future.done():
            future.add_done_callback(cls._consume_future_exception)
        executor.shutdown(wait=future.done())

SqliteDurability

async_batch_llm.SqliteDurability

Bases: str, Enum

SQLite synchronization policy for committed artifact batches.

ResultSerializationError

async_batch_llm.ResultSerializationError

Bases: ValueError, AsyncBatchLLMError

A result value or serialized payload is not safely supported.

Artifact errors

async_batch_llm.ArtifactError

Bases: RuntimeError, AsyncBatchLLMError

Base class for artifact preparation, format, and persistence failures.

async_batch_llm.ArtifactIdentityError

Bases: ArtifactError

One item's strategy cannot satisfy the store's identity contract.

async_batch_llm.ArtifactSerializationError

Bases: ArtifactError

An artifact identity/input/result could not be canonically serialized.

async_batch_llm.ArtifactIOError

Bases: ArtifactError

An artifact could not be read, written, flushed, or closed.

async_batch_llm.ArtifactFormatError

Bases: ArtifactError

An artifact is malformed or uses an unsupported schema version.