Python SDK Reference

from bitfab import (
    BITFAB_PROGRESS_PREFIX,
    AdaptContext,
    AddDatasetGradersResult,
    AddDatasetTracesResult,
    AllowedEnvVars,
    Bitfab,
    BitfabFunction,
    BitfabClaudeAgentHandler,
    BitfabLangChainCallbackHandler,
    BitfabLangGraphCallbackHandler,
    BitfabLangGraphIntegration,
    BitfabOpenAIAgentHandler,
    BitfabTracingProcessor,
    CaptureWhen,
    CapturedSpan,
    CodeChangeFile,
    ConcurrencyPrimitive,
    CurrentSpan,
    CurrentTrace,
    Dataset,
    DatasetGraderRef,
    DatasetTraceIds,
    DatasetsClient,
    DetachedTrace,
    MixedTracingError,
    DbBranchLease,
    DbBranchOptions,
    DbBranchReplayError,
    DbSnapshotRef,
    GraderRerun,
    GraderRerunProgress,
    GraderRerunResult,
    GraderRerunStatus,
    MockOverride,
    MockOverrideCtx,
    MockOverrideInput,
    MockOverrideResolver,
    MockValue,
    NO_MOCK_OVERRIDE,
    NodeMatcher,
    OutputTarget,
    RemoveDatasetGradersResult,
    RemoveDatasetTracesResult,
    ReplayBranch,
    ReplayConcurrency,
    ReplayError,
    ReplayExperimentStart,
    ReplayInterrupt,
    ReplayInterruptSignal,
    ReplayItem,
    ReplayItemFinishEvent,
    ReplayItemStartProgress,
    ReplayItemFinishProgress,
    ReplayProgress,
    ReplayRegistration,
    ReplayRegistry,
    ReplayRegistryContext,
    ReplayResult,
    RerunGradersResult,
    SaveAssertion,
    SaveDatasetResult,
    SpanNodeMeta,
    SpanOccurrence,
    SpanListField,
    SpanListResult,
    SpanListType,
    SpanTarget,
    SpanType,
    SpansClient,
    TraceAssertion,
    TraceAssertionSource,
    TraceAssertionsResult,
    TraceAssertionsUpdate,
    TraceTarget,
    TraceTargetOccurrence,
    TracesClient,
    finalizers,               # finalizers.openai_chunks, finalizers.anthropic_events
    flush_traces,
    get_current_replay_branch,
    get_current_span,
    get_current_trace,
    report_replay_progress,
    seed_from_registry,
    serialize_replay_result,
)

SpanType = Literal["llm", "agent", "function", "guardrail", "handoff", "custom"]
CaptureWhen = Literal["always", "nested"]
ConcurrencyPrimitive = Literal["async", "process"]
MockStrategy = Literal["none", "all", "marked"]

Bitfab(
    api_key: str | Callable[[], str | None] | None = None,
    service_url: Optional[str] = None,
    env_vars: Optional[AllowedEnvVars] = None,
    capture_enabled: bool | None = None,
    baml_client: Any = None,
    strict: bool = False,
    trace_across_threads: bool | None = None,
    max_captured_subtree_spans: int | None = None,
    simulation_plan: bool = True,
    enabled: bool | None = None,
)

client.api_key: str | None
client.capture_enabled: bool
client.enabled: bool  # deprecated alias

def trace(
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    mock_on_replay_default: bool = False,
    max_depth: Optional[int] = None,
    max_captured_subtree_spans: Optional[int] = None,
    exclude: Collection[str] = (),
    include_wrappers: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def span(
    trace_function_key: str,
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    capture_when: CaptureWhen = "always",
    test_run_id: Optional[str] = None,  # deprecated alias for experiment_id
    mock_on_replay: bool = False,
    finalize: Optional[Callable[[Any], Any]] = None,
    experiment_id: Optional[str] = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def trace(
    trace_function_key: str,
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    mock_on_replay_default: bool = False,
    max_depth: Optional[int] = None,
    max_captured_subtree_spans: Optional[int] = None,
    exclude: Collection[str] = (),
    include_wrappers: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def node(
    *,
    name: Optional[str] = None,
    type: Optional[SpanType] = None,
    capture: bool = True,
    test_run_id: Optional[str] = None,  # deprecated alias for experiment_id
    mock_on_replay: Optional[bool] = None,
    finalize: Optional[Callable[[Any], Any]] = None,
    variant: Optional[Union[str, Callable[..., Optional[str]]]] = None,
    experiment_id: Optional[str] = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def get_function(trace_function_key: str) -> BitfabFunction

# Form 1: uses baml_client from constructor
def wrap_baml(
    method: Callable[..., Any],
    *,
    on_collector: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

# Form 2: explicit client
def wrap_baml(
    baml_client: Any,
    method: Callable[..., Any],
    *,
    on_collector: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def wrap_jev(
    fn: Callable[..., Any],
    *,
    on_reply: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def wrap_openrouter(
    fn: Callable[..., Any],
    *,
    on_response: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def wrap_anthropic(
    fn: Callable[..., Any],
    *,
    on_response: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def wrap_openai_responses(
    fn: Callable[..., Any],
    *,
    on_response: Callable[[Any], None] | None = None,
) -> Callable[..., Any]

def get_trace(trace_id: str) -> DetachedTrace

def get_trace_span(
    trace_id: str,
    *,
    id: Optional[str] = None,
    name: Optional[str] = None,
    type: Optional[SpanType] = None,
    occurrence: Literal["first", "last"] | int = "last",
) -> Optional[CapturedSpan]

first_model_call = client.get_trace_span(trace_id, type="llm", occurrence="first")

def get_span(span_id: str) -> Optional[CapturedSpan]

client.datasets.save(trace_function_key: str, name: str, description: str | None = None, *, golden: bool | None = None) -> SaveDatasetResult
client.datasets.list(trace_function_key: str | None = None) -> list[Dataset]
client.datasets.get(dataset_id: str) -> Dataset
client.datasets.update(dataset_id: str, *, name: str | None = None, description: str | None = None, golden: bool | None = None) -> Dataset
client.datasets.list_traces(dataset_id: str) -> DatasetTraceIds
client.datasets.add_traces(dataset_id: str, trace_ids: list[str]) -> AddDatasetTracesResult
client.datasets.remove_traces(dataset_id: str, trace_ids: list[str]) -> RemoveDatasetTracesResult
client.datasets.add_graders(dataset_id: str, grader_ids: list[str]) -> AddDatasetGradersResult
client.datasets.remove_graders(dataset_id: str, grader_ids: list[str]) -> RemoveDatasetGradersResult
client.datasets.rerun_graders(dataset_id: str, grader_ids: list[str] | None = None, *, timeout: float = 90.0) -> RerunGradersResult
await client.datasets.rerun_graders_async(dataset_id: str, grader_ids: list[str] | None = None, *, timeout: float = 90.0) -> RerunGradersResult
client.datasets.get_grader_rerun(dataset_id: str, run_id: str | None = None) -> GraderRerun | None

class DatasetGraderRef(TypedDict):
    id: str
    name: str | None

class Dataset(TypedDict):
    id: str
    traceFunctionKey: str
    name: str
    description: str | None
    traceCount: int
    graders: list[DatasetGraderRef]
    golden: bool
    createdAt: str
    updatedAt: str

class SaveDatasetResult(TypedDict):
    dataset: Dataset
    created: bool

class DatasetTraceIds(TypedDict):
    datasetId: str
    traceIds: list[str]

class AddDatasetTracesResult(TypedDict):
    dataset: Dataset
    addedTraceIds: list[str]
    alreadyPresentTraceIds: list[str]
    skippedTraceIds: list[str]

class RemoveDatasetTracesResult(TypedDict):
    dataset: Dataset
    removedTraceIds: list[str]
    notPresentTraceIds: list[str]

class AddDatasetGradersResult(TypedDict):
    dataset: Dataset
    addedGraderIds: list[str]
    alreadyAssignedGraderIds: list[str]
    skippedGraderIds: list[str]

class RemoveDatasetGradersResult(TypedDict):
    dataset: Dataset
    removedGraderIds: list[str]
    notAssignedGraderIds: list[str]

GraderRerunStatus = Literal["pending", "running", "completed", "errored"]

class GraderRerun(TypedDict):
    id: str
    status: GraderRerunStatus
    graderIds: list[str]
    progress: GraderRerunProgress | None
    result: GraderRerunResult | None
    error: str | None
    createdAt: str
    updatedAt: str

class RerunGradersResult(TypedDict):
    run: GraderRerun
    joinedExisting: bool

client.traces.search(*, caller_metadata: dict[str, str] | None = None, name: str | None = None, name_contains: str | None = None, experiment_id: str | None = None, trace_function_key: str | None = None, cursor: str | None = None, limit: int | None = None) -> TraceSearchResult
client.traces.get_assertions(trace_id: str) -> TraceAssertionsResult
client.traces.save_assertions(trace_id: str, assertions: list[SaveAssertion], source: TraceAssertionSource = "agent", *, skip_guided_rewrite: bool = False) -> list[SavedTraceAssertion]
client.traces.save_assertions_all(updates: list[TraceAssertionsUpdate], source: TraceAssertionSource = "agent", *, skip_guided_rewrite: bool = False) -> list[SavedTraceAssertion]
client.traces.archive_assertions(trace_id: str, assertion_ids: list[str]) -> list[str]
client.traces.generate_assertions(trace_id: str, *, guidance: str | None = None, category_ids: list[str] | None = None, regenerate: bool | None = None, timeout: float = 600.0) -> list[TraceAssertion]
await client.traces.generate_assertions_async(trace_id: str, *, guidance: str | None = None, category_ids: list[str] | None = None, regenerate: bool | None = None, timeout: float = 600.0) -> list[TraceAssertion]
client.traces.get_scenario(trace_id: str) -> Scenario
client.traces.save_scenario(trace_id: str, *, name: str | None = None, description: str | None = None, trigger: ScenarioTrigger | None = None, data: ScenarioData | None = None, append_data: ScenarioDataAppend | None = None, actual_outcome: ScenarioActualOutcome | None = ..., category_ids: list[str] | None = None) -> Scenario

saved = client.traces.save_assertions_all(
    [
        {"traceId": trace_id, "assertions": [{"assertion": "Departs before 9am"}]}
        for trace_id in trace_ids
    ]
)

drafts = client.traces.generate_assertions(
    trace_id,
    guidance="The agent booked a flight after the user asked for a train.",
)

async def generate_for_all(trace_ids: list[str]):
    return await asyncio.gather(
        *(client.traces.generate_assertions_async(t) for t in trace_ids)
    )

TraceTargetOccurrence = Literal["first", "last"] | int

TraceAssertionSource = Literal["human", "agent"]

class OutputTarget(TypedDict):
    kind: Literal["output"]

class SpanTarget(TypedDict):
    kind: Literal["span"]
    name: str
    occurrence: NotRequired[TraceTargetOccurrence]

TraceTarget = OutputTarget | SpanTarget

class SaveAssertion(TypedDict):
    category_id: NotRequired[str | None]
    id: NotRequired[str]
    assertion: str
    humanNote: NotRequired[str | None]
    passCriteria: NotRequired[str | None]
    failCriteria: NotRequired[str | None]
    targetOnEvaluatedTrace: NotRequired[TraceTarget | None]

class TraceAssertion(TypedDict):
    category_id: str | None
    category: CategorySummary | None
    id: str
    traceId: str
    assertion: str
    humanNote: str | None
    passCriteria: str | None
    failCriteria: str | None
    targetOnEvaluatedTrace: TraceTarget | None
    source: TraceAssertionSource
    createdAt: str
    updatedAt: str

AssertionGenerationStatus = Literal["running", "completed", "failed"]

class AssertionGeneration(TypedDict):
    status: AssertionGenerationStatus
    error: str | None
    assertionIds: list[str]
    startedAt: str
    finishedAt: str | None

class TraceAssertionsResult(TypedDict):
    assertions: list[TraceAssertion]
    inheritedFrom: str | None
    generation: AssertionGeneration | None

class TraceAssertionsUpdate(TypedDict):
    traceId: str
    assertions: list[SaveAssertion]

client.traces.save_scenario(
    trace_id,
    append_data={"missing": ["The traveler's loyalty program number"]},
)

client.traces.save_scenario(
    trace_id,
    actual_outcome={
        "summary": "Booked the 9am flight but never emailed the confirmation",
        "actions": [
            {
                "name": "book_flight",
                "details": {"destination": "DEN"},
                "result": "Confirmation ABC123",
                "failed": False,
                "at": None,
            },
        ],
    },
)

client.traces.save_scenario(trace_id, actual_outcome=None)

client.traces.save_scenario(trace_id, category_ids=[category["id"]])

client.traces.save_scenario(trace_id, category_ids=[])

class ScenarioSource(TypedDict):
    spanId: str
    sourceField: Literal["input", "output"]
    text: str

ScenarioTrigger = TypedDict(
    "ScenarioTrigger",
    {
        "inputId": int | None,
        "inputType": str,
        "channel": str,
        "from": str,
        "to": str | None,
        "cc": str | None,
        "subject": str | None,
        "type": str,
        "eventTime": str,
        "eventTimeSource": Literal["message date", "run start"],
        "source": Literal["request", "run instructions"],
        "content": str,
        "contentPassages": list[str],
        "contentIsWholeMessage": bool,
        "sources": list[ScenarioSource],
    },
)

ScenarioDataOrigin = TypedDict(
    "ScenarioDataOrigin",
    {
        "section": str,
        "channel": str | None,
        "from": str | None,
        "to": str | None,
        "cc": str | None,
        "subject": str | None,
        "sentAtText": str | None,
    },
)

class ScenarioHistoryEntry(TypedDict):
    role: Literal["thread start", "fact"]
    kind: str
    quote: str
    summary: str | None
    origin: ScenarioDataOrigin | None
    sentAt: str | None
    weekday: str | None
    sincePrevious: str | None
    source: NotRequired[ScenarioSource | None]

class ScenarioData(TypedDict):
    hasHistory: bool
    topic: str
    history: list[ScenarioHistoryEntry]
    missing: list[str]

class ScenarioDataAppend(TypedDict, total=False):
    history: list[ScenarioHistoryEntry]
    missing: list[str]

class ScenarioAction(TypedDict):
    name: str
    details: dict[str, Any]
    result: str | None
    failed: bool
    at: str | None

class ScenarioActualOutcome(TypedDict):
    summary: str
    actions: list[ScenarioAction]
    source: NotRequired[ScenarioSource | None]

class Scenario(TypedDict):
    id: str
    createdAt: str
    updatedAt: str
    organizationId: str
    traceFunctionId: str
    sourceTraceId: str
    extractor: Literal["noah"]
    name: str
    description: str
    trigger: ScenarioTrigger
    data: ScenarioData
    actualOutcome: ScenarioActualOutcome | None
    categories: list[CategorySummary]
    triggerSummary: str
    dataSummary: str

client.spans.list(trace_id: str, *, span_type=None, span_name=None, fields=None, cursor=None, limit=None) -> SpanListResult

page = client.spans.list(trace_id, fields=["input"])
for span in page["spans"]:
    print(span["type"], span["name"], span["input"])

client.organization_members.list() -> list[OrganizationMember]

class OrganizationMember(TypedDict):
    id: str
    fullName: str | None
    email: str | None
    imageUrl: str | None

client.categories.save(title: str, *, id: str | None = None, description: str | None = None) -> Category
client.categories.get(id: str) -> Category
client.categories.list() -> list[Category]
client.categories.delete(id: str) -> Category

class CategorySummary(TypedDict):
    id: str
    title: str
    description: str

class JustificationSpan(TypedDict):
    spanId: str
    text: str

Justification = list[JustificationSpan]

class Category(CategorySummary):
    organizationId: str
    createdAt: str
    updatedAt: str

category = client.categories.save(
    "Mandate compliance", description="Checks explicit user instructions"
)
assertion = client.traces.save_assertions(
    trace_id,
    [{"assertion": "Escalates conflicting calendar requests", "category_id": category["id"]}],
)[0]
client.traces.save_assertions(
    trace_id,
    [{"id": assertion["id"], "assertion": assertion["assertion"], "category_id": None}],
)
client.categories.delete(category["id"])

client.verdicts.save(verdict: Verdict, annotation: str | None = None, trace_id: str | None = None, original_trace_id: str | None = None, attempt: int | None = None, confidence: VerdictConfidence | None = None, test_run_id: str | None = None, assertion_id: str | None = None, evidence: Justification | None = None, experiment_id: str | None = None) -> VerdictOutcome
client.verdicts.save_all(verdicts: list[VerdictUpdate], test_run_id: str | None = None, experiment_id: str | None = None) -> list[VerdictOutcome]
client.verdicts.skip(trace_id: str | None = None, original_trace_id: str | None = None, attempt: int | None = None, test_run_id: str | None = None, assertion_id: str | None = None, experiment_id: str | None = None, annotation: str | None = None) -> VerdictOutcome
client.verdicts.abstain(trace_id: str | None = None, original_trace_id: str | None = None, attempt: int | None = None, test_run_id: str | None = None, assertion_id: str | None = None, experiment_id: str | None = None, annotation: str | None = None) -> VerdictOutcome
client.verdicts.archive(trace_id: str | None = None, original_trace_id: str | None = None, attempt: int | None = None, test_run_id: str | None = None, assertion_id: str | None = None, experiment_id: str | None = None) -> VerdictOutcome
client.verdicts.save_human(verdict: Literal["pass", "fail"], annotation: str, trace_id: str, confidence: VerdictConfidence | None = None, assertion_id: str | None = None, evidence: Justification | None = None) -> HumanVerdictOutcome
client.verdicts.save_human_all(verdicts: list[HumanVerdictUpdate]) -> list[HumanVerdictOutcome]
client.verdicts.get(trace_id: str) -> TraceVerdicts | None
client.verdicts.get_all(trace_ids: list[str]) -> list[TraceVerdicts]
client.verdicts.generate_evidence(trace_id: str, assertion_id: str) -> list[AssertionVerdictEvidence]

VerdictConfidence = Literal["VeryLow", "Low", "Medium", "High", "VeryHigh"]
VerdictAction = Literal["set", "archived", "no-active-verdict", "skipped", "abstained"]
Verdict = Literal["pass", "fail", "abstained"]

class VerdictUpdate(TypedDict):
    traceId: NotRequired[str]
    originalTraceId: NotRequired[str]
    attempt: NotRequired[int]
    assertionId: NotRequired[str]
    verdict: NotRequired[Verdict]
    annotation: NotRequired[str]
    confidence: NotRequired[VerdictConfidence]
    evidence: NotRequired[Justification | None]
    skip: NotRequired[bool]
    archive: NotRequired[bool]

class VerdictOutcome(TypedDict):
    key: str
    traceId: str
    action: VerdictAction

class HumanVerdictUpdate(TypedDict):
    traceId: str
    assertionId: str
    verdict: Literal["pass", "fail"]
    annotation: str
    confidence: NotRequired[VerdictConfidence]
    evidence: NotRequired[Justification | None]

class HumanVerdictOutcome(TypedDict):
    traceId: str
    assertionId: str | None
    verdict: Literal["pass", "fail"]
    action: Literal["set"]

VerdictStatus = Literal["judged", "skipped", "not-judged"]
VerdictSource = Literal["human", "agent"]

class AssertionVerdict(TypedDict):
    assertionId: str
    assertion: str | None
    verdict: Verdict | None
    verdictStatus: VerdictStatus
    annotation: str | None
    confidence: VerdictConfidence | None
    source: VerdictSource
    approved: bool

class TraceVerdicts(TypedDict):
    traceId: str
    verdict: Verdict | None
    verdictStatus: VerdictStatus
    annotation: str | None
    approved: bool
    passed: int
    failed: int
    abstained: int
    assertions: list[AssertionVerdict]

client.graders.get_verdicts(trace_ids: list[str] | None = None, grader_id: str | None = None, limit: int | None = None) -> list[GraderVerdict]

GraderVerdictSource = Literal["human", "live_grader"]

class GraderVerdict(TypedDict):
    traceId: str
    graderId: str
    graderName: str | None
    graderStatus: GraderStatus
    graderType: str
    verdict: Literal["pass", "fail"] | None
    reason: str | None
    failureDiagnostic: str | None
    confidence: VerdictConfidence | None
    source: GraderVerdictSource
    evaluatedAt: str | None

result = client.traces.get_assertions(item.original_trace_id)

for assertion in result["assertions"]:
    if assertion["approvalState"] != "approved":
        continue
    judgement = judge(assertion, replay_output)
    if judgement is None:
        client.verdicts.skip(
            original_trace_id=item.original_trace_id,
            attempt=item.attempt,
            assertion_id=assertion["id"],
            experiment_id=run.experiment_id,
        )
    else:
        client.verdicts.save(
            verdict="pass" if judgement.passed else "fail",
            annotation=judgement.summary,
            original_trace_id=item.original_trace_id,
            attempt=item.attempt,
            assertion_id=assertion["id"],
            experiment_id=run.experiment_id,
        )

client.graders.save_verdict(grader_id: str, trace_id: str, verdict: Literal["pass", "fail"], reason: str | None = None, failure_diagnostic: str | None = None, confidence: VerdictConfidence | None = None) -> GraderVerdictOutcome
client.graders.save_verdict_all(verdicts: list[GraderVerdictUpdate]) -> list[GraderVerdictOutcome]

class GraderVerdictUpdate(TypedDict):
    graderId: str
    traceId: str
    verdict: Literal["pass", "fail"]
    reason: NotRequired[str]
    failureDiagnostic: NotRequired[str]
    confidence: NotRequired[VerdictConfidence]

class GraderVerdictOutcome(TypedDict):
    graderId: str
    traceId: str
    verdict: Literal["pass", "fail"]
    action: Literal["set"]

client.experiments.list(*, dataset_id=None, trace_function_key=None, git_branch=None, experiment_group_id=None, status=None, metadata=None, created_after=None, created_before=None, cursor=None, limit=None) -> ExperimentPage
client.experiments.get(id: str) -> Experiment
client.experiments.get_rollup(id: str) -> ExperimentRollupResult
client.experiments.get_rollup_all(ids: list[str]) -> list[ExperimentRollupResult]

registry = ReplayRegistry()
registry.register("checkout", bitfab, run_checkout)
registry.register(
    "agent",
    bitfab,
    run_agent,
    trace_function_key="support-agent",
    mock="marked",
    adapt_inputs=adapt_inputs,
    options_factory=lambda ctx: {
        "mock_override": create_support_agent_mock(ctx.params.get("scenario"))
    },
)

class ReplayRegistry:
    @property
    def names(self) -> tuple[str, ...]: ...  # registered command names, in registration order

    def register(
        self,
        name: str,
        client: Bitfab,
        fn: Callable[..., Any],
        *,
        trace_function_key: str | None = None,
        options_factory: Callable[[ReplayRegistryContext], dict[str, Any]] | None = None,
        **options: Any,
    ) -> ReplayRegistry: ...

    def get(self, name: str) -> ReplayRegistration: ...

def register_mock_override(
    match_or_override: str | NodeMatcher | MockOverride | MockOverrideResolver,
    value: Any = ...,
) -> None

def clear_mock_overrides() -> None

def replay(
    fn_or_key: Callable | str | None = None,
    fn: Optional[Callable] = None,  # required when fn_or_key is a key string, or when fn_or_key is omitted (legacy replay(fn=decorated_fn))
    *,
    limit: Optional[int] = None,  # default 5; max 5,000; ignored with trace_ids/datasets
    trace_ids: Optional[list[str]] = None,  # max 100; with a dataset selector it pins which members replay
    name: str | None = None,  # what this run is testing, such as 'baseline'; commit, branch, and datasets are already recorded
    notes: str | None = None,  # run conditions Bitfab cannot see, such as a forced feature flag (max 10,000 chars)
    metadata: dict[str, str] | None = None,  # your own tags on the experiment; keys match ^[A-Za-z0-9_.:-]{1,40}$, values at most 500 chars, at most 50 keys
    max_concurrency: Optional[int] = 10,
    attempts: int = 1,  # 1-100; replay each trace this many times, attempt-major
    concurrency: ReplayConcurrency | None = None,  # supersedes attempts + max_concurrency
    code_change_description: str | None = ...,
    code_change_files: list[CodeChangeFile] | None = ...,  # omit for rename-aware git diff of uncommitted changes vs HEAD (vs the branch it was cut from when clean); None disables capture
    mock: MockStrategy = "marked",
    experimental_selective_replay: SelectiveReplayOptions | None = None,
    mock_override: MockOverrideInput | list[MockOverrideInput] | None = None,
    adapt_inputs: Callable[[list[Any], dict[str, Any], AdaptContext], tuple[list[Any], dict[str, Any]]] | None = None,
    experiment_group_id: str | None = None,  # joins that group; an ID no group has yet creates it
    experiment_group_name: str | None = None,  # names the group when this replay creates it or it has no name; requires experiment_group_id
    db_branch: DbBranchOptions | bool | None = None,  # True requests a DB branch per replay item
    dry_run: bool = False,  # resolve every item's inputs and call nothing
    dataset_id: str | None = None,  # durably attributes the experiment to a dataset
    dataset_ids: list[str] | None = None,  # several datasets: replays their union, attributed to each
    grader_ids: list[str] | None = None,  # graders attached to this run, unioned with the dataset's at grading (max 100)
    only_with_assertions: bool = False,  # narrow the selection to traces carrying an approved assertion
    skip_assertion_judging: bool = False,
    on_item_start: Callable[[ReplayItemStartProgress], None] | None = None,
    on_item_finish: Callable[[ReplayItemFinishProgress], None] | None = None,
    on_progress: Callable[[ReplayProgress], None] | None = None,  # deprecated compatibility callback
    on_experiment_start: Callable[[ReplayExperimentStart], None] | None = None,
    resume: str | None = None,  # experiment ID to pick back up, running only its unfinished trace and attempt pairs
    force: bool = False,  # with resume only: skip the recent-activity and code-change refusals
    on_interrupt: Callable[[ReplayInterrupt], None] | None = None,  # on SIGINT or SIGTERM, once the experiment is marked interrupted
    process_launcher: ProcessLauncher | None = None,  # internal: wired by bitfab-replay --registry for primitive="process"
) -> ReplayResult

class ReplayExperimentStart(TypedDict):
    experiment_id: str
    experiment_url: str

ReplayInterruptSignal = Literal["SIGINT", "SIGTERM"]

class ReplayInterrupt(TypedDict):
    experiment_id: str
    experiment_url: str
    signal: ReplayInterruptSignal
    marked_interrupted: bool

def seed_trace(
    trace_function_key: str,
    fn: Callable,
    *,
    args: Sequence[Any] = (),
    kwargs: Mapping[str, Any] | None = None,
    metadata: Mapping[str, Any] | None = None,
    session_id: str | None = None,
    name: str | None = None,
) -> str

bitfab = Bitfab(capture_enabled=False)

trace_id = bitfab.seed_trace(
    "agent-turn",
    run_ticket,
    kwargs={"ticket_id": "T-1"},
    metadata={"case_uid": "c-1", "suite": "smoke"},
)

bitfab.replay(run_ticket, trace_ids=[trace_id], adapt_inputs=lambda args, kwargs, ctx: (args, kwargs))

def reseed_trace(
    trace_function_key: str,
    fn: Callable,
    *,
    trace_id: str,
) -> ReseedResult

def reseed_from_registry(
    registry: ReplayRegistry,
    pipeline: str,
    trace_ids: Sequence[str],
) -> dict[str, Any]

def seed_from_registry(
    registry: ReplayRegistry,
    pipeline: str,
    cases: Sequence[Mapping[str, Any]],
) -> dict[str, Any]

def call(method_name: str, **kwargs: Any) -> Any

def get_langgraph_callback_handler(trace_function_key: str) -> BitfabLangGraphCallbackHandler

def get_langgraph_integration(
    trace_function_key: str,
    *,
    mock_tools_on_replay: bool | list[str] | tuple[str, ...] = True,
) -> BitfabLangGraphIntegration

integration.middleware -> BitfabAgentMiddleware
integration.instrument(graph: GraphT) -> GraphT  # a traced copy of the same type
integration.create_agent(*args, **kwargs) -> CompiledStateGraph
integration.create_deep_agent(*args, **kwargs) -> CompiledStateGraph

from langgraph.prebuilt import ToolNode

integration = bitfab.get_langgraph_integration("support-agent")
tools = [lookup_customer, search_docs]
tool_node = ToolNode(
    tools,
    wrap_tool_call=integration.wrap_tool_call,
    awrap_tool_call=integration.awrap_tool_call,
)
support_graph = integration.instrument(
    build_support_graph(tools=tools, tool_node=tool_node)
)

result = support_graph.invoke(
    {"messages": [{"role": "user", "content": "Help with order 123"}]},
    config={"configurable": {"thread_id": "customer-456"}},
)

bitfab.replay("support-agent", support_graph.invoke, limit=10)

def get_deep_agents_integration(
    trace_function_key: str,
    *,
    mock_tools_on_replay: bool | list[str] | tuple[str, ...] = True,
) -> BitfabLangGraphIntegration

integration = bitfab.get_deep_agents_integration("research-agent")
agent = integration.create_deep_agent(
    model="openai:gpt-4o-mini",
    tools=[internet_search],
    subagents=[research_subagent],
)

for chunk in agent.stream({"messages": [{"role": "user", "content": "What is LangGraph?"}]}):
    ...

bitfab.replay("research-agent", agent.invoke, limit=10)

def get_openai_tracing_processor() -> BitfabOpenAITracingProcessor

def get_openai_agent_handler(trace_function_key: str) -> BitfabOpenAIAgentHandler

def get_claude_agent_handler(trace_function_key: str) -> BitfabClaudeAgentHandler

def span(
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    capture_when: CaptureWhen = "always",
    test_run_id: Optional[str] = None,  # deprecated alias for experiment_id
    mock_on_replay: bool = False,
    finalize: Optional[Callable[[Any], Any]] = None,
    experiment_id: Optional[str] = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def trace(
    *,
    name: Optional[str] = None,
    type: SpanType = "custom",
    mock_on_replay_default: bool = False,
    max_depth: Optional[int] = None,
    max_captured_subtree_spans: Optional[int] = None,
    exclude: Collection[str] = (),
    include_wrappers: bool = False,
) -> Callable[[Callable[P, T]], Callable[P, T]]

def get_claude_agent_handler() -> BitfabClaudeAgentHandler

def get_langgraph_callback_handler() -> BitfabLangGraphCallbackHandler

def get_langchain_callback_handler() -> BitfabLangGraphCallbackHandler

def get_langgraph_integration(
    *,
    mock_tools_on_replay: bool | list[str] | tuple[str, ...] = True,
) -> BitfabLangGraphIntegration

def get_deep_agents_integration(
    *,
    mock_tools_on_replay: bool | list[str] | tuple[str, ...] = True,
) -> BitfabLangGraphIntegration

def wrap_baml(
    method_or_client: Any,
    method: Optional[Callable] = None,
) -> Callable[..., Any]

def wrap_openrouter(
    fn: Callable[..., Any],
    *,
    on_response: Optional[Callable[[Any], None]] = None,
) -> Callable[..., Any]

def wrap_anthropic(
    fn: Callable[..., Any],
    *,
    on_response: Optional[Callable[[Any], None]] = None,
) -> Callable[..., Any]

def wrap_openai_responses(
    fn: Callable[..., Any],
    *,
    on_response: Optional[Callable[[Any], None]] = None,
) -> Callable[..., Any]

def finalizers.openai_chunks(chunks: list[Any]) -> dict[str, Any]
def finalizers.anthropic_events(events: list[Any]) -> dict[str, Any]

def get_current_span() -> CurrentSpan | _NoOpCurrentSpan