Worker Job Adapter

Use this tutorial when provider work happens inside background jobs and you do not want to add handle.capture(...) inside every job body.

ReplayLab still works the same way:

1. initialize the SDK once when the worker starts
2. keep normal provider code inside the job
3. decorate the job entrypoint with capture_job(...)
4. replay the captured provider capsule locally

The decorator opens one job-scoped capture for each function call. Jobs that do not call supported providers do not write capsules by default.

Setup

From the repo checkout, install the development environment:

uv sync --all-packages --all-groups

Run the maintained worker scenario:

python scripts/run_scenario.py run job-worker-local --keep-workspace

Expected ending:

ReplayLab scenario passed.
Scenario: job-worker-local
Tier: loopback
Boundaries: 1
Payloads: 2
Providers: requests

This means ReplayLab captured one provider boundary during a decorated job, stopped the provider server, replayed the same job command without that provider, compared the report, generated pytest, and ran the generated test.

For a richer maintainer check of sync, async, and provider-free jobs, run:

python scripts/run_scenario.py run job-lifecycle-local --keep-workspace

That scenario invokes one provider-free decorated job, one sync provider job, and one async provider job. It expects the provider-free job to write no capsule, then replays and generates pytest regressions for the two provider-backed job capsules.

Decorate A Job

Initialize ReplayLab near worker startup:

import replaylab
from replaylab import CapturePayloadPolicy

handle = replaylab.init(
    project_name="support-worker",
    auto_patch_integrations=("requests",),
    capture_payload_policy=CapturePayloadPolicy.FULL,
)

Decorate the job function:

import requests
from replaylab.integrations.jobs import capture_job


@capture_job(
    handle=handle,
    name="sync_ticket",
    session_id_arg="ticket_id",
    queue_name="support",
    worker_name="worker-a",
)
def sync_ticket(ticket_id: str) -> dict[str, str]:
    response = requests.get(
        f"https://support.example.test/tickets/{ticket_id}",
        timeout=5,
    )
    response.raise_for_status()
    return {"ticket_id": ticket_id}

You do not call handle.capture(...) inside the function. The decorator creates the capture scope and the provider wrappers attach boundaries to it.

What Gets Captured

ReplayLab records provider boundaries and safe job metadata:

• job name
• callable module and qualname
• optional queue name
• optional worker name
• optional session_id_arg name
• the extracted session ID on the run when session_id_arg is configured

ReplayLab does not record job args, kwargs, return values, queue payloads, provider payload bodies, or secrets at the job adapter layer. Provider request and response payload capture still follows the configured ReplayLab payload policy and redaction policy.

Inspect The Capsule

After a job that calls a supported provider, list capsules:

uv run replaylab capsule list --local-store-root .replaylab

Look for a capsule with integrations like:

requests, job, auto_patch, same_process

Inspect it:

uv run replaylab capsule inspect <capsule_id> --local-store-root .replaylab

You should see one HTTP boundary attached to the job step.

Replay

Replay runs the same worker command under ReplayLab's replay runtime:

uv run replaylab replay <capsule_id> \
  --local-store-root .replaylab \
  --auto-patch-integrations requests \
  --report-id replay_sync_ticket \
  -- python worker.py

Compare the report:

uv run replaylab report compare \
  <capsule_id> \
  .replaylab/replays/replay_sync_ticket/report.json \
  --local-store-root .replaylab

Expected result:

Status: succeeded
Boundaries: expected=1, replayed=1, problems=0

This means the provider call was served from the capsule rather than from the live dependency.

Generate A Regression

Generate a pytest provider replay guard from the capsule:

uv run replaylab generate-test <capsule_id> \
  --output tests/regression/test_sync_ticket_replay.py \
  --fixture-root tests/fixtures/replaylab/capsules \
  --app-root . \
  --auto-patch-integrations requests \
  -- python worker.py

Run it:

uv run pytest tests/regression/test_sync_ticket_replay.py

The generated test runs your worker command through replaylab replay, asserts the replay report, and fails if the job no longer matches the captured provider boundary.

Current Boundaries

• The adapter is a framework-agnostic decorator, not a Celery, RQ, or APScheduler integration.
• Jobs without provider boundaries do not write capsules unless write_empty_captures=True.
• Job args, kwargs, return values, and queue payloads are not captured by the job adapter.
• Replay still uses replaylab replay for local regression execution.
• Cloud upload, hosted issue grouping, and parent/child capsule merging are future work.