Review sync

The review-sync subsystem turns completed items from a Langfuse annotation queue into curated records in a golden evaluation dataset. A human reviewer examines each trace or span, optionally edits the expected output, and marks the item complete in the Langfuse UI. The review-sync service then reads those completed items, applies a configurable promotion policy, maps fields using declarative projection rules, and writes the approved records into the target golden dataset.

Pods interact with the subsystem in one of two ways:

Path	Covered in	When to use
Declarative bindings	This file	Register a `QueueToDatasetBinding` with the `review_sync_app` service, then trigger a sync job whenever reviews should be promoted (`eval-review sync`); the service reads the queue, checks the policy and publishes, with no pod code.
Pod-side helpers	06-review-sync-pod-helpers.md	Call `LangfuseReviewArtifactReader` + `build_promoted_records()` + `publish_promoted_records()` directly from pod code for custom promotion flows or one-off scripts.

Both paths use the same shared contracts and produce records in the same EvaluationItem shape consumed by offline evaluation.

See glossary.md for term definitions. All types described here are exported from semantic_evaluation.orchestration.review_sync.

Flow overview

graph TD
    AQ["Langfuse annotation queue\nreviewed items COMPLETED"]
    RS["review_sync_app\ncentralized service"]
    BIND["QueueToDatasetBinding\ndeclarative recipe"]
    POLICY["ReviewPolicy\nwhat counts as promotable"]
    PROJ["Projection and ValueSelector\nfield mapping rules"]
    GD["Langfuse golden dataset\ndev or release"]
    ARTS["Local run artifacts\nreviewed records, summary, checkpoint"]
    NEXT["Offline evaluation\nCI gates and dashboards"]

    BIND --> RS
    POLICY --> BIND
    PROJ --> BIND
    AQ -- "LangfuseReviewArtifactReader\nreads and normalizes" --> RS
    RS -- "promotion policy check" --> RS
    RS -- "publish_promoted_records" --> GD
    RS -- "persist_review_sync_artifacts" --> ARTS
    GD --> NEXT

    style AQ fill:#efe9ed,stroke:#62294b,stroke-width:2px
    style RS fill:#e7eeec,stroke:#125a40,stroke-width:2px
    style GD fill:#efe9ed,stroke:#62294b,stroke-width:2px
    style NEXT fill:#e5f4fd,stroke:#0098eb,stroke-width:2px
    style BIND fill:#e9ebec,stroke:#273943,stroke-width:2px
    style POLICY fill:#e9ebec,stroke:#273943,stroke-width:2px
    style PROJ fill:#e9ebec,stroke:#273943,stroke-width:2px
    style ARTS fill:#e5f4fd,stroke:#0098eb,stroke-width:2px

Setting up review in Langfuse

Before a binding can promote anything, the Langfuse project needs the queue the reviewers work in and the scores they give. Both are made once per scope, in the Langfuse UI (Human Annotation → Annotation Queues → New Queue, and Manage Score Configs from there) or through the Langfuse API (langfuse.api.score_configs.create(...), langfuse.api.annotation_queues.create_queue(...)).

One queue per scope. A queue item is one trace or one observation; the binding projects that object's input and output into a dataset item. Queue the observation whose input and output are the scope's (the intent step's span for an intent scope, the root trace for an end-to-end scope), so the projection reads the right fields and the scores belong to one scope. A trace that an evaluation run also touched still reads as its root observation: the run's sdk-experiment span is ignored. The review_sync_app service reads annotation and correction scores through Langfuse's v3 scores API, so it must run a release that includes this reader.
A score config per score the policy requires. The names must equal review_policy.required_score_names exactly; a queue item without every one of them is not promoted. NUMERIC from 1 to 5 matches the policy's default bounds. Attach the configs to the queue, so the reviewer sees them.
A corrected output is what the reviewer writes when the agent's output was wrong (Langfuse's "corrected output" on the trace or observation, stored as a score of data type CORRECTION). The policy requires one whenever a score is at or below low_score_correction_threshold. In a projection, selected_output is the corrected output where there is one and the original output otherwise.
The queue id is the first id in the queue's URL, .../annotation-queues/<QUEUE_ID>/items/<ITEM_ID>, or the id the API returns. The binding's queue_id is that one, never an item's.

Items reach the queue by hand (Add to queue on a trace or observation) or through the API (langfuse.api.annotation_queues.create_queue_item(...)). A sync reads only the items a reviewer has marked completed, and only their scores given in the queue (source ANNOTATION).

What a selector's queue_item_path can read, for each reviewed item:

Path	Holds
`original_input`, `original_output`	The queued trace's or observation's input and output, as objects
`selected_output`	The corrected output, or the original output when there is none
`corrected_output`	The corrected output only; empty when the reviewer wrote none
`review_scores.<name>`	One score the reviewer gave, with its `value`
`record_id`, `trace_id`, `observation_id`, `queue_item_id`, `reviewed_at`	The item's identity; `record_id` is the queued object's id, and becomes the dataset item's id
`review_metadata`	The queue item's details, `selected_agent` and `workflow_version` read from the trace, and the raw payloads under `raw_artifacts`

Declarative bindings

`QueueToDatasetBinding`

A binding is the pod's declarative recipe for one (annotation queue, golden dataset, scope) triple. Registering a binding with the service is the only pod-side configuration step needed for the fully-managed path.

Field	Type	Required	Description
`binding_id`	`str`	yes	Stable unique identifier for this binding
`queue_id`	`str`	yes	Langfuse annotation queue ID to read from
`langfuse_profile_id`	`str`	yes	Langfuse connection profile registered in the service (supplies project credentials for this queue)
`dataset_scope`	`str`	yes	Evaluation scope name written to `EvaluationItem.metadata.evaluation_scope`
`product_id`	`str`	yes	Short product identifier written to `EvaluationItem.metadata.product_id`
`dataset_version`	`str`	yes	Version label stamped on promoted records
`created_by`	`str`	yes	Author tag written to record metadata
`review_policy`	`ReviewPolicy`	yes	When a reviewed record is considered promotable
`dataset_target`	`DatasetTarget`	yes	Target Langfuse dataset definition
`input_projection`	`Projection`	yes	How to map queue-item fields to `EvaluationItem.input`
`expected_output_projection`	`Projection`	yes	How to map reviewer output to `EvaluationItem.expected_output`
`description`	`str \\| None`	no	Human-readable description of this binding
`queue_cleanup`	`QueueCleanupConfig`	no	Whether to delete queue items after promotion (default: disabled)
`metadata_mapping`	`MetadataMapping`	no	Additional metadata to write to each promoted record

`ReviewPolicy`

Controls which reviewed records are eligible for promotion. See also glossary.md — ReviewPolicy.

Field	Type	Default	Description
`required_score_names`	`list[str]`	— (required)	Score names that must be present on the queue item
`normalized_score_min`	`float`	`1.0`	Minimum accepted score
`normalized_score_max`	`float`	`5.0`	Maximum accepted score
`low_score_correction_threshold`	`float`	`3.0`	Score at or below this value requires a corrected output from the reviewer

A record is promotable when every score in required_score_names is present and within [normalized_score_min, normalized_score_max]. When any score falls at or below low_score_correction_threshold, the reviewer must supply an explicit corrected output — records approved as is are not promotable in that case.

`DatasetTarget`

Describes the Langfuse golden dataset to write promoted records into.

Field	Type	Required	Description
`name`	`str`	yes	Langfuse dataset name (follows the naming convention)
`description`	`str`	yes	Human-readable description written to the Langfuse dataset
`metadata`	`dict`	no	Free-form metadata written to the Langfuse dataset object
`input_schema`	`dict`	yes	JSON Schema object describing the `input` field shape
`expected_output_schema`	`dict`	yes	JSON Schema object describing the `expected_output` field shape

Projections, fields, and selectors

Three types together describe how to read values out of a completed queue item and write them into an EvaluationItem:

Projection — describes one side (input or expected output) of the resulting dataset record. It holds an ordered list of ProjectionField entries.

ProjectionField — one key on the resulting record:

Field	Description
`dataset_field`	The key written into the record (e.g. `"query"` on the input side)
`selectors`	Ordered fallback chain of `ValueSelector` entries — the first non-null value wins
`required`	Whether a null result blocks the record (default: `True`)
`default`	Fallback value when `required=False` and all selectors yield null

ValueSelector — one value source. Exactly one of these must be set:

Field	Description
`queue_item_path`	Dotted path into the queue-item payload (e.g. `"selected_output.reply_text"`)
`literal`	A constant value injected regardless of the queue-item payload
`coerce`	`ValueCoercion` applied after reading — `identity`, `string`, `boolean`, `integer`, or `number` (default: `identity`)

ValueCoercion handles the fact that Langfuse stores annotation-queue payloads loosely (a reviewer may type a number as a string). The coercion runs at the boundary so downstream dataset code can rely on strict types.

Example — binding that maps a trace output field with a literal fallback:

from semantic_evaluation.orchestration.review_sync import (
    QueueToDatasetBinding, ReviewPolicy, DatasetTarget,
    Projection, ProjectionField, ValueSelector,
)

binding = QueueToDatasetBinding(
    binding_id="pathfinder-routing-v2",
    queue_id="<langfuse-queue-id>",
    langfuse_profile_id="main",
    dataset_scope="routing",
    product_id="pathfinder",
    dataset_version="1.6",
    created_by="team-ai-evaluation",
    review_policy=ReviewPolicy(required_score_names=["correctness", "completeness"]),
    dataset_target=DatasetTarget(
        name="pathfinder_agent_routing_dev",
        description="Pathfinder routing ground truth from annotation queue review",
        input_schema={"type": "object"},
        expected_output_schema={"type": "object"},
    ),
    input_projection=Projection(fields=[
        ProjectionField(
            dataset_field="messages",
            selectors=[ValueSelector(queue_item_path="original_input.messages")],
        ),
    ]),
    expected_output_projection=Projection(fields=[
        ProjectionField(
            dataset_field="expected_mode",
            selectors=[
                ValueSelector(queue_item_path="selected_output.expected_mode"),
                ValueSelector(literal="generic"),
            ],
            required=False,
            default="generic",
        ),
    ]),
)

`MetadataMapping` and `ScalarMapping`

MetadataMapping controls additional metadata written to each promoted record.

Field	Type	Default	Description
`record_id`	`ScalarMapping`	`queue_item_path="record_id"`	Stable record identifier used as the Langfuse dataset item ID
`workflow_version`	`ScalarMapping`	empty (not written)	Workflow version that produced the trace
`selected_agent`	`ScalarMapping`	empty (not written)	Agent variant — useful when multiple variants run in parallel (A/B comparisons)
`interaction_type`	`str`	`"annotation_queue_review"`	Interaction type tag written to record metadata
`tags`	`list[str]`	`[]`	Category labels added to the record; they become trace tags on the prediction traces the item produces

ScalarMapping holds an ordered list of ValueSelector entries plus an optional default. The first non-null selector result wins.

`QueueCleanupConfig`

Field	Type	Default	Description
`enabled`	`bool`	`False`	When `True`, the service deletes completed queue items from Langfuse after successful promotion so reviewers see a clean queue. Keep disabled during early-stage development to allow replay.

Managing bindings — `ReviewSyncServiceClient`

ReviewSyncServiceClient is the HTTP client pods use to register bindings and trigger sync jobs against the review_sync_app service. See glossary.md — ReviewSyncServiceClient.

from semantic_evaluation.orchestration.review_sync import ReviewSyncServiceClient

client = ReviewSyncServiceClient(
    base_url="http://review-sync-service:8002",
    timeout_seconds=120.0,
    api_key=None,  # your pod's caller key, sent as X-API-Key; omit or None to send no header
)

base_url is the service host only. The client adds the API major (/v1) to every route, so a product never configures a version; see service API versions.

api_key is your pod's own caller key for the review-sync service. The deployed service requires a known key on the /v1/langfuse-profiles routes, reads included, so the two profile methods need api_key. Bindings and sync jobs are in a warning phase: a call with a known key is identified in the service logs, and a call without one is still accepted until an enforcement date announced in the review-sync service's changelog, so pass your key to every call.

Method	Returns	Description
`list_bindings()`	`list[QueueToDatasetBinding]`	Return all registered bindings
`get_binding(binding_id)`	`QueueToDatasetBinding`	Fetch one binding by ID
`upsert_binding(binding)`	`QueueToDatasetBinding`	Create or fully replace a binding
`validate_binding(binding)`	`BindingValidationResult`	Validate a binding definition without registering it
`run_sync(request)`	`JobResult`	Trigger a sync job for one binding and return the result
`get_job(job_id)`	`JobResult`	Load a previously stored job result
`register_langfuse_profile(profile_id=, base_url=, public_key=, secret_key=)`	`dict`	Register or replace the Langfuse project a binding reads from; returned without its secret key. Needs `api_key`
`list_langfuse_profiles()`	`list[dict]`	Every registered profile, without secret keys. Needs `api_key`

All methods are async. Raises RuntimeError on non-2xx responses.

The `eval-review` command

A project needs no code for review: it keeps each binding as a file and uses eval-review to register it and to run its syncs. The command talks to REVIEW_SYNC_SERVICE_BASE_URL (default http://127.0.0.1:8002) and, like eval-dataset, reads ./.env and ./.env.local first.

eval-review register-profile --profile-id <product>          # once per Langfuse project; needs your caller key as REVIEW_SYNC_API_KEY
eval-review profiles                                         # what is registered, without secret keys
eval-review validate evaluation/review/end_to_end.binding.yaml
eval-review register-binding evaluation/review/end_to_end.binding.yaml   # validates, then registers
eval-review bindings
eval-review sync --file evaluation/review/end_to_end.binding.yaml --dry-run
eval-review sync --file evaluation/review/end_to_end.binding.yaml        # or --binding-id <id>

register-profile sends LANGFUSE_BASE_URL, LANGFUSE_PUBLIC_KEY and LANGFUSE_SECRET_KEY from the environment, and prints the stored profile without the secret. Every binding that reads a queue in that project names the profile in langfuse_profile_id.
register-binding registers nothing when the service rejects the binding, and exits 1. Re-registering a binding_id replaces the binding.
sync --dry-run reports what would be promoted and writes nothing. sync exits 1 when the job failed, with its error.
Each prints the service's answer as JSON. A file that is not a valid binding, or a service that cannot be reached, exits 2 with the reason.

A binding file is a QueueToDatasetBinding in YAML (or JSON):

binding_id: refunds-end_to_end
description: Promote reviewed refund-agent turns into the end-to-end dev dataset.
queue_id: <queue id>
langfuse_profile_id: refunds
dataset_scope: end_to_end
product_id: refunds
dataset_version: dev
created_by: refunds team
review_policy:
  required_score_names: [Correct tools, Correct reply]
dataset_target:
  name: refunds_agent_end_to_end_dev
  description: The refund agent's end-to-end items.
  input_schema:
    type: object
    required: [message]
    properties: {message: {type: string}}
  expected_output_schema:
    type: object
    required: [response]
    properties: {response: {type: string}}
input_projection:
  fields:
    - dataset_field: message
      selectors: [{queue_item_path: original_input.message, coerce: string}]
expected_output_projection:
  fields:
    - dataset_field: response
      selectors: [{queue_item_path: selected_output.response, coerce: string}]

The dataset_target names the scope's dev dataset, the one the evaluation manifest's dataset_sync.scope_to_dataset gives for that scope, so reviewed items join the items curated there and are released with them (03-datasets-lifecycle.md). The projections write the dataset item's fields; the schemas must declare every dataset_field the projections write, which the service checks when it validates.

Job execution

`JobRequest`

Field	Type	Default	Description
`binding_id`	`str`	required	Which binding to run
`dry_run`	`bool`	`False`	Report what would be written without writing
`max_items`	`int \\| None`	`None`	Cap the number of items processed in this job
`page_size`	`int \\| None`	`None`	Langfuse queue fetch page size
`persist_artifacts`	`bool`	`True`	Write local run artifacts to `outputs_dir`

`JobResult` and `JobSummary`

JobResult is the stored and returned record for one completed job run.

Field	Description
`job_id`	Unique job identifier
`status`	`JobStatus.COMPLETED` or `JobStatus.FAILED`
`started_at`, `finished_at`	ISO 8601 UTC timestamps
`request`	The `JobRequest` that triggered this job
`summary`	`JobSummary` with item counts and outcomes (present on `COMPLETED` jobs)
`error`	Error message (present on `FAILED` jobs)

JobSummary item-count fields:

Field	Description
`completed_reviewed_items`	Items read from the queue with status COMPLETED
`already_handled_items`	Items skipped because already in the checkpoint
`promotable_items`	Items that passed the `ReviewPolicy` check
`approved_as_is_items`	Items promoted without reviewer correction
`explicit_correction_items`	Items promoted with reviewer-supplied corrections
`published_items`	Items successfully written to the golden dataset
`skipped_unchanged_items`	Items already in the dataset with identical content
`policy_skipped_items`	Items that did not meet the review policy
`failed_publish_items`	Items where the Langfuse write failed
`dry_run`	Whether this was a dry-run job

Outcome types

`HandledRecordOutcome`

A record that did not need a publish write — either already handled or policy-skipped. Carries record_id, queue_item_id, dataset_scope, dataset_name, reason (POLICY_SKIPPED or SKIPPED_UNCHANGED), and an optional message.

`BlockedRecordOutcome`

A record that stopped further processing — typically a failed publish. Carries the same fields as HandledRecordOutcome plus reason (FAILED_PUBLISH).

`RecordPublishOutcome` and `PublishResult`

RecordPublishOutcome carries the result of one Langfuse write attempt: exactly one of published=True, skipped_unchanged=True, or error (a string) is set.

PublishResult aggregates a list of RecordPublishOutcome entries and exposes three computed properties: published_items, skipped_unchanged_items, and failed_publish_items.

Checkpointing

The service tracks which (record_id, reviewed_at) pairs have been handled in a CheckpointState stored per binding. On the next sync run, already-handled versions are skipped, so it is safe to run sync jobs repeatedly without double-promoting records.

CheckpointState carries:

Field	Description
`binding_id`	Which binding this checkpoint belongs to
`handled_versions`	List of `HandledVersion` (record_id + queue_item_id + reviewed_at)
`pending_cleanup_items`	Queue items marked for deletion retry
`updated_at`, `updated_by_job_id`	Audit fields

CheckpointAdvance is the delta recorded for one completed job — the new HandledVersion entries to merge into the checkpoint.

When a sync fails

Symptom	Cause	What to do
`404 Annotation queue not found`	One message, three causes: the queue item id was used instead of the queue id (both are in `.../annotation-queues/<QUEUE_ID>/items/<ITEM_ID>`); the binding names the wrong Langfuse profile, so the service looks in a project without that queue; or the queue was deleted and recreated, which gives it a new id, and the binding still has the old one	Check them in that order. Re-registering the binding with the current queue id replaces the old one
`401` from `register-profile` or `profiles`	`REVIEW_SYNC_API_KEY` is missing, or is not a known caller key	`eval-docs show access-and-credentials`
The job fails at credential resolution	The binding's `langfuse_profile_id` is not registered, or was deleted	`eval-review profiles`, then `register-profile`
`completed_reviewed_items` is 0	No item in the queue is marked completed	Reviewers complete items in the queue; scores alone are not enough
`policy_skipped_items` is above 0	A required score is missing or out of bounds, or a score at or below `low_score_correction_threshold` has no corrected output	The outcome's `message` names which; the score config names must equal `required_score_names` exactly
`already_handled_items` covers everything	Each reviewed version is promoted once (see Checkpointing)	Nothing: a new review of the item is a new version and is promoted on the next sync

`default_dataset_name`

A naming-convention helper exported alongside the binding types:

from semantic_evaluation.orchestration.review_sync import default_dataset_name

# With schema type:  "{product_id}_{schema_type}_{scope}_v{version}"
name = default_dataset_name("pathfinder", "routing", "1.6", schema_type="agent")
# → "pathfinder_agent_routing_v1.6"

# Without schema type: "{product_id}_{scope}_v{version}"
name = default_dataset_name("pathfinder", "routing", "1.6")
# → "pathfinder_routing_v1.6"

See glossary.md and 03-datasets-lifecycle.md — Naming convention for the full naming convention.

Key types reference

Type	Source file	Purpose
`QueueToDatasetBinding`	`service_contracts.py`	Declarative recipe: queue to dataset
`ReviewPolicy`	`service_contracts.py`	What counts as a promotable reviewed record
`DatasetTarget`	`service_contracts.py`	Target Langfuse dataset definition
`Projection`	`service_contracts.py`	Shape contract for one side of a promoted record
`ProjectionField`	`service_contracts.py`	One field in a projection with a fallback selector chain
`ValueSelector`	`service_contracts.py`	One value source: queue-item path or literal constant
`ValueCoercion`	`service_contracts.py`	Coercion enum: `identity`, `string`, `boolean`, `integer`, `number`
`MetadataMapping`	`service_contracts.py`	Additional metadata written to each promoted record
`ScalarMapping`	`service_contracts.py`	One scalar value with a selector chain
`QueueCleanupConfig`	`service_contracts.py`	Whether to delete queue items after promotion
`JobRequest`	`service_contracts.py`	Parameters for one sync job execution
`JobResult`	`service_contracts.py`	Stored result of one completed sync job
`JobStatus`	`service_contracts.py`	`COMPLETED` or `FAILED`
`JobSummary`	`service_contracts.py`	Item-count breakdown for a completed job
`HandledRecordOutcome`	`service_contracts.py`	Record skipped without a publish write
`BlockedRecordOutcome`	`service_contracts.py`	Record that stopped further processing
`RecordPublishOutcome`	`service_contracts.py`	One Langfuse write attempt result
`PublishResult`	`service_contracts.py`	Aggregate publication result
`CheckpointState`	`service_contracts.py`	Service-owned handled-version state per binding
`CheckpointAdvance`	`service_contracts.py`	Checkpoint delta from one completed job
`BindingListResponse`	`service_contracts.py`	Response wrapper for the binding list endpoint
`BindingValidationResult`	`service_contracts.py`	Result of binding pre-validation
`ReviewSyncServiceClient`	`client.py`	HTTP client for the `review_sync_app` service
`LangfuseReviewArtifactReader`	`langfuse.py`	Reads and normalizes completed annotation queue items — see pod helpers
`ReviewArtifactReader`	`contracts.py`	Protocol for pluggable artifact readers — see pod helpers
`ReviewedObservationRecord`	`contracts.py`	Transport-neutral reviewed artifact representation — see pod helpers
`PromotedRecordMapping`	`contracts.py`	Caller-provided mapping from reviewed record to dataset record — see pod helpers
`ReviewTransport`	`contracts.py`	Enum of supported artifact-reading transport modes
`build_promoted_records`	`promotion.py`	Convert reviewed artifacts to `EvalCaseRecord` objects — see pod helpers
`publish_promoted_records`	`publisher.py`	Write `EvalCaseRecord` objects to Langfuse — see pod helpers
`persist_review_sync_artifacts`	`artifacts.py`	Write local JSON artifact files for a sync run — see pod helpers
`default_dataset_name`	`service_contracts.py`	Naming-convention helper

Review sync

Flow overview

Setting up review in Langfuse

Declarative bindings

QueueToDatasetBinding

ReviewPolicy

DatasetTarget

Projections, fields, and selectors

MetadataMapping and ScalarMapping

QueueCleanupConfig

Managing bindings — ReviewSyncServiceClient

The eval-review command