Skip to content

Sync execution

Definition: sync job A sync job is one immediate execution of the review-sync service for one stored review binding. It reads reviewed items, attempts publication, and stores a machine-readable result. See sync job.

The JobRequest/JobResult/JobSummary contract shape is owned by sdk/semantic_evaluation and documented in 06-review-sync.md — Job execution. This page describes what review_sync_service/executor.py actually does when POST /v1/sync-jobs runs one job.

Execution flow

graph TD
  A["POST /v1/sync-jobs"] --> B["1. Validate binding"]
  B --> C["2. Resolve Langfuse credentials"]
  C --> D["3. Read completed queue items"]
  D --> E["4. Filter already-handled versions"]
  E --> F["5. Sort pending records"]
  F --> G["6. Evaluate each record"]
  G --> H["7. Queue cleanup"]
  H --> I["8. Advance checkpoint"]
  I --> J["9. Persist artifacts"]
  J --> K["10. Persist job record"]
  K --> L["Return JobResult"]
  style A fill:#e5f4fd,stroke:#0098eb,stroke-width:2px
  style L fill:#e5f4fd,stroke:#0098eb,stroke-width:2px

For each POST /v1/sync-jobs request (JobRequest: binding_id, dry_run, max_items, page_size, persist_artifacts):

  1. Validate the binding. The same semantic checks as POST /v1/bindings/validate (see 01-bindings.md — Binding validation) run again before anything is read from Langfuse. A binding that fails validation here fails the job with status: failed.
  2. Resolve Langfuse credentials. The executor looks up binding.langfuse_profile_id in the Langfuse profile registry. An unregistered profile fails the job immediately with a message telling the caller to register it first.
  3. Read completed queue items. The shared LangfuseReviewArtifactReader reads and normalizes completed items from binding.queue_id, honoring request.page_size (falling back to the service's REVIEW_SYNC_QUEUE_PAGE_SIZE default) and request.max_items.
  4. Filter already-handled versions. Records whose (queue_item_id, record_id, reviewed_at) identity is already present in this binding's checkpoint state are dropped before anything else runs — this is what makes repeated sync jobs safe to run without double-promoting.
  5. Sort the remaining pending records by (reviewed_at, queue_item_id, trace/observation identity) for a stable, deterministic processing order.
  6. Evaluate each pending record in order:
  7. If the review policy rejects it, record a policy_skipped outcome and continue to the next record.
  8. Otherwise, map it into a promoted record (see Reviewer comments and score passthrough below) and count it as approved_as_is or explicit_correction depending on whether the reviewer supplied a correction.
  9. On a dry run, stop here — mark the record handled without publishing.
  10. Otherwise, publish it. A skipped_unchanged outcome is non-blocking and processing continues. A failed_publish outcome is blocking: the job stops immediately, and every later pending record in that job becomes deferred (attempted again on the next sync job for this binding, since it was never marked handled).
  11. Resolve queue cleanup (see 03-langfuse-profiles-and-cleanup.md — Queue cleanup).
  12. Advance the checkpoint if this was not a dry run and something actually changed (new handled records or a change in pending cleanup retries).
  13. Persist artifacts (see Artifacts below) if persist_artifacts is true (the default).
  14. Persist the job record to REVIEW_SYNC_JOBS_DIR/{job_id}.json, regardless of outcome, so GET /v1/sync-jobs/{job_id} can return it later.

Promotion outcomes

Outcome Blocking? Meaning
published no The record was written to the golden dataset for the first time (or with changed content)
skipped_unchanged no The record already exists in the golden dataset with identical content — see publish_promoted_records()
policy_skipped no The record did not meet the review policy
failed_publish yes The Langfuse write failed; the job stops and every later pending record becomes deferred
deferred — Not an outcome value itself — a pending record that was never reached because an earlier record in the same job failed to publish 
graph TD
  A["pending record"] --> B{"review policy passes?"}
  B -->|"no"| C["policy_skipped\nnon-blocking — continue"]
  B -->|"yes"| D["map to promoted record"]
  D --> E{"dry run?"}
  E -->|"yes"| F["mark handled\nno publish — continue"]
  E -->|"no"| G["publish to golden dataset"]
  G --> H{"result"}
  H -->|"write succeeded"| I["published\nnon-blocking — continue"]
  H -->|"content unchanged"| J["skipped_unchanged\nnon-blocking — continue"]
  H -->|"write failed"| K["failed_publish\nBLOCKING"]
  K --> L["remaining records deferred\njob stops"]
  style C fill:#e7eeec,stroke:#55aa77,stroke-width:2px
  style F fill:#e7eeec,stroke:#55aa77,stroke-width:2px
  style I fill:#e7eeec,stroke:#55aa77,stroke-width:2px
  style J fill:#e7eeec,stroke:#55aa77,stroke-width:2px
  style K fill:#fdecea,stroke:#cc4444,stroke-width:2px
  style L fill:#fdecea,stroke:#cc4444,stroke-width:2px

policy_skipped and skipped_unchanged are reported as HandledRecordOutcome entries; failed_publish is reported as a single BlockedRecordOutcome (at most one per job, since it stops processing). See 06-review-sync.md — Outcome types for the exact field shapes.

Reviewer comments and score passthrough

Reviewer comments and human review scores are direct passthrough, not binding-configurable projections:

  • Reviewer comments copy unchanged from the reviewed record's reviewer_comments list into the promoted record's metadata.
  • Human review scores copy unchanged from the reviewed record's review_scores dict into the promoted record's metadata as human_review_scores.

Both live inside EvalCaseRecord.metadata, so they also participate in the unchanged-content comparison publish_promoted_records() runs before writing — if only a reviewer's comment changed between two reviews of the same record, that is enough to make the record published again rather than skipped_unchanged.

Checkpointing

Checkpoint state (contract shape documented in 06-review-sync.md — Checkpointing) is stored per binding by checkpoint_store.py as one JSON file per binding_id. Two things are specific to how this service maintains it:

  • Advancement is conditional. A completed job only writes a new CheckpointState when it is not a dry run and either it handled at least one record or its pending cleanup-retry set changed. A job that reads the queue and finds nothing new to do leaves the checkpoint untouched.
  • Cleanup retry state travels with the checkpoint. pending_cleanup_items carries queue items whose deletion failed on the last attempt so a later job can retry them — see 03-langfuse-profiles-and-cleanup.md.

Artifacts

When persist_artifacts is true (the default), the executor writes JSON artifacts under REVIEW_SYNC_ARTIFACTS_DIR/{binding_id}/{job_id}/. Each artifact type is written twice per job: as a last_*.json file (always overwritten, for inspecting the most recent run) and as a timestamped *_{timestamp}.json file (for historical comparison). The full artifact list — reviewed, pending, handled, blocked, deferred, and promoted records plus cleanup candidates, cleanup outcomes, cleanup summary, sync summary, and before/after checkpoint snapshots — is documented in Pod-side helpers: persist_review_sync_artifacts(). Promoted-record artifacts include reviewer comments and human review scores inside record metadata whenever the reviewed record carried them.

Reset local artifacts and state with:

uv run python scripts/reset_local_state.py