Jobs and workers¶

PostProject persists the production meaning of requested work. Your host, service, or script remains responsible for choosing jobs, starting processes, managing resources, and applying retry policy.

The protocol between a host that wants work and a worker that does it:

        sequenceDiagram
    participant Host as Requesting host
    participant P as Production
    participant Worker
    Host->>P: request a job: inputs, output target, parameters
    Worker->>P: list requested jobs of a kind it understands
    Worker->>P: claim with tool identity and duration
    P-->>Worker: owning scoped lease
    loop while the work outlasts the lease
        Worker->>P: renew with the lease
    end
    alt the tool succeeds
        Worker->>P: one transaction: stage the output,<br/>record its activity, complete with the lease
    else the tool fails
        Worker->>P: fail with the lease and a bounded diagnostic
    end
    P-->>Host: revision with JobSucceeded or JobFailed

Each arrow into the production is a committed transaction; the sections below describe them in order.

Claim, renew, and release¶

A worker claims a requested job with tool attribution, optional agent attribution, and a duration. The returned owning lease is pending until its transaction commits. Claiming and completing within that same transaction is supported. Claim before staging the output when using native bindings.

Renew before expiry, giving a duration that extends the current expiry. Release returns the job to requested state without recording failure:

static pp_error_code_t claim_renew_release(pp_production_t *production,
                                           pp_job_id_t job_id,
                                           pp_error_t **error) {
  pp_transaction_t *transaction = NULL;
  pp_job_lease_t *lease = NULL, *imported = NULL;
  char *token = NULL;
  pp_error_code_t status =
      pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  if (status == PP_OK) {
    status = pp_transaction_claim_job_lease(
        transaction, job_id, "Example Transcoder", "3.2",
        "https://example.com/transcoder", "render-node-04", "com.example.host",
        "node-04", NULL, 5 * MINUTE, &lease, error);
  }
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }
  pp_transaction_release(transaction);
  transaction = NULL;

  pp_production_id_t scope;
  pp_job_id_t scoped_job;
  uint32_t state = 0;
  int64_t expiry = 0;
  if (status == PP_OK) {
    status = pp_job_lease_get(lease, &scope, &scoped_job, &state, &expiry, error);
  }
  if (status == PP_OK && (state != PP_JOB_LEASE_ACTIVE || expiry <= 0 ||
      memcmp(&scoped_job, &job_id, sizeof job_id) != 0)) status = PP_ERROR_INTERNAL;

  pp_job_set_t *claimed = NULL;
  pp_job_claim_t detail = {0};
  if (status == PP_OK) status = pp_production_job(production, job_id, &claimed, error);
  if (status == PP_OK) status = pp_job_set_get_claim(claimed, 0, &detail, error);
  if (status == PP_OK && strcmp(detail.tool_name, "Example Transcoder") != 0) status = PP_ERROR_INTERNAL;
  pp_job_set_release(claimed);

  /* Explicit private transport for another process; never print the token. */
  if (status == PP_OK) status = pp_job_lease_export_token(lease, &token, error);
  if (status == PP_OK) {
    status = pp_production_import_job_lease(production, (const uint8_t *)token,
                                          strlen(token), &imported, error);
  }
  pp_string_release(token);
  if (status == PP_OK) status = pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  if (status == PP_OK) status = pp_transaction_renew_job_lease(transaction, imported, 10 * MINUTE, error);
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }
  pp_transaction_release(transaction);
  transaction = NULL;

  /* Releasing abandons the work without recording a failure. */
  if (status == PP_OK) status = pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  if (status == PP_OK) status = pp_transaction_release_job_lease(transaction, imported, error);
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }
  pp_job_lease_free(imported);
  pp_job_lease_free(lease);
  pp_transaction_release(transaction);
  return status;
}

C++

void claim_renew_release(postproject::Production &production,
                         const postproject::JobId &job_id) {
  auto claim = production.edit().value();
  auto lease = claim.claimJobLease(job_id, encoder, std::chrono::minutes(1), worker).value();
  require(std::holds_alternative<postproject::PendingJobLease>(lease.state().value()), "pending lease");
  claim.commit().value();

  require(lease.productionId().value() == production.id().value() && lease.jobId().value() == job_id,
          "lease scope");
  require(std::holds_alternative<postproject::ActiveJobLease>(lease.state().value()), "active lease");
  // Export only for private worker transport, never ordinary output.
  auto imported = production.importJobLease(lease.exportToken().value()).value();
  auto renew = production.edit().value();
  renew.renewJobLease(imported, std::chrono::minutes(2)).value();
  renew.commit().value();

  auto release = production.edit().value();
  release.releaseJobLease(imported).value(); // back to requested
  release.commit().value();
  require(std::holds_alternative<postproject::ClosedJobLease>(imported.state().value()), "closed lease");
  imported.close();
}

Python

def claim_renew_release(production: Production, job_id: JobId) -> None:
    with production.edit() as transaction:
        lease = transaction.claim_job_lease(
            job_id, WORKER, timedelta(minutes=5), agent=AGENT
        )
        assert isinstance(lease.state, PendingJobLease)
        transaction.commit()

    with lease:
        assert lease.production_id == production.id and lease.job_id == job_id
        assert isinstance(lease.state, ActiveJobLease)
        # Explicit private transport when work moves to another process.
        with production.import_job_lease(lease.export_token()) as imported:
            with production.edit() as transaction:
                transaction.renew_job_lease(imported, timedelta(minutes=10))
                transaction.commit()
            with production.edit() as transaction:
                transaction.release_job_lease(imported)
                transaction.commit()
            assert isinstance(imported.state, ClosedJobLease)

Rust

fn claim_renew_release(production: &mut SqliteProduction, job_id: JobId) -> Result<()> {
    let tool = ToolIdentity::new("Example Worker", Some("1.0".to_owned()), None)?;
    let agent = AgentIdentity::new(Some("Render node 4".to_owned()), None)?;

    let mut transaction = production.begin_transaction()?;
    let lease =
        transaction.claim_job_lease(job_id, &tool, Some(&agent), Duration::from_secs(60))?;
    assert_eq!(lease.state()?, JobLeaseState::Pending);
    transaction.commit()?;
    drop(transaction);
    assert_eq!(lease.production_id(), production.production().id());
    assert_eq!(lease.job_id(), job_id);
    assert!(matches!(lease.state()?, JobLeaseState::Active { .. }));

    // Explicit private transport when another process takes over this worker.
    let imported = production.import_job_lease(&lease.export_token()?)?;
    let mut transaction = production.begin_transaction()?;
    transaction.renew_job_lease(&imported, Duration::from_secs(120))?;
    transaction.release_job_lease(&imported)?;
    transaction.commit()?;
    assert_eq!(imported.state()?, JobLeaseState::Closed);
    Ok(())
}

CLI

CLAIM=$(postproject --json job claim jobs.pproj "$JOB_ID" \
  --tool-name "Proxy Worker" --tool-version 1.0 --agent-name worker-1 \
  --lease 5m --lease-token-file claim.token)
# Credentials stay in the private file, separate from ordinary job facts.
postproject job renew jobs.pproj "$JOB_ID" --lease-token-file claim.token --lease 10m
postproject job release jobs.pproj "$JOB_ID" --lease-token-file claim.token

Every worker transition checks production, job, current credential and expiry under the writer lock, including a final expiry check before commit. Failed commits close newly pending leases and roll back publication. Dropping a lease only frees local resources; it never releases or renews a job. Coordinator cancellation invalidates an active lease immediately.

Use Rust Duration, integral C++ std::chrono::duration, Python timedelta, or C’s explicitly named microsecond scalar. Durations must be positive exact whole microseconds, at most 24 hours. Caller-selected current times are not accepted. Storage preserves a clock high-water mark across reopen; backward jumps reject guarded transitions until time catches up, and forward jumps may expire work.

Export a token explicitly to pass a lease between processes. Import validates its version, size, production, job and current authority state. Tokens grant local worker authority and must stay private; they do not authenticate a user. The CLI creates a new --lease-token-file FILE for a claim and reads that file for later transitions. Use --lease-token-file - to read a token from stdin. Existing output files are never overwritten. New Unix files have mode 0600; Windows files inherit the directory’s ACL. Choose a private directory.

CLI --lease values are an integer followed by us, ms, s, m or h, with 60s as the default. Claim reserves the output file before writing production state. If delivery fails after commit, the error retains the committed receipt; the claim remains durable until expiry or cancellation. An incomplete owned file is removed. Listings, JSON results, events and diagnostics never print tokens.

Complete a job¶

On success, stage the output representation and the activity that produced it, then complete the job in that same transaction. The output, the activity with its input snapshots, and the job’s transition to succeeded become durable together or not at all:

static pp_error_code_t
complete_proxy(pp_production_t *production, pp_job_id_t job_id,
               const pp_representation_id_t *input_id, const pp_asset_id_t *asset_id,
               const char *output_path, pp_representation_id_t *out_proxy_id,
               pp_error_t **error) {
  const int64_t started = NOW + 10 * MINUTE;
  const int64_t finished = NOW + 12 * MINUTE;
  pp_transaction_t *transaction = NULL;
  pp_job_lease_t *lease = NULL;
  pp_activity_id_t activity_id = {{0}};
  pp_error_code_t status =
      pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  if (status == PP_OK) {
    status = pp_transaction_claim_job_lease(transaction, job_id, "Example Transcoder",
                                      "3.2", "https://example.com/transcoder",
                                      NULL, NULL, NULL, NULL, 5 * MINUTE, &lease, error);
  }
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }
  pp_transaction_release(transaction);
  transaction = NULL;
  if (status == PP_OK && !run_transcoder(output_path)) {
    status = PP_ERROR_IO;
  }

  /* Stage the output and the activity, then complete in the same
   * transaction; never commit them earlier. */
  if (status == PP_OK) {
    status = pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  }
  if (status == PP_OK) {
    status = pp_transaction_add_file_representation(transaction, *asset_id, PP_REPRESENTATION_PROXY, output_path, out_proxy_id, error);
  }
  if (status == PP_OK) {
    const pp_activity_edge_t inputs[] = {{*input_id, NULL}};
    const pp_activity_edge_t outputs[] = {{*out_proxy_id, NULL}};
    status = pp_transaction_create_activity(transaction,
        &(pp_activity_recipe_t){PROXY_JOB, inputs, 1, &started, &finished, "Example Transcoder", "3.2", "https://example.com/transcoder", NULL, NULL, NULL, NULL, NULL, 0},
        outputs, 1, &activity_id, error);
  }
  if (status == PP_OK) {
    status =
        pp_transaction_complete_job_lease(transaction, lease,
                                    *out_proxy_id, activity_id, error);
  }
  if (status == PP_OK) {
    /* Parameters on the activity make the artifact reproducible. */
    pp_object_ref_t activity;
    status = pp_object_ref_from_activity(activity_id, &activity, error);
    if (status == PP_OK) status = stage_proxy_parameters(transaction, &activity, error);
  }
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }
  pp_job_lease_free(lease);
  pp_transaction_release(transaction);
  return status;
}

C++

postproject::RepresentationId complete_proxy(postproject::Production &production,
                                 const postproject::Job &job,
                                 const std::string &output_path) {
  auto claim = production.edit().value();
  auto lease = claim.claimJobLease(job.id, encoder, std::chrono::minutes(10), worker).value();
  claim.commit().value();

  std::ofstream(output_path, std::ios::binary) << "encoded proxy";

  // Output, provenance, and the job transition commit together or not at all.
  auto transaction = production.edit().value();
  const auto proxy_id = transaction.addRepresentation(
      job.output_asset_id, job.output_representation_kind, output_path).value();
  postproject::ActivitySpec activity{};
  activity.kind = job.kind;
  for (const auto &input : job.inputs) {
    activity.inputs.push_back(input);
  }
  activity.outputs = {proxy_id};
  activity.tool = encoder;
  activity.agent = worker;
  const auto activity_id = transaction.createActivity(activity).value();
  // Parameters on the activity let the artifact be regenerated later.
  transaction.addMetadataValue(
      postproject::ObjectRef::activity(activity_id), transcode, "profile",
      postproject::MetadataValue::plainString("editing-proxy")).value();
  transaction.completeJobLease(lease, proxy_id, activity_id).value();
  transaction.commit().value();
  return proxy_id;
}

Python

def run_proxy_job(production: Production, job_id: JobId, output: Path) -> None:
    with production.edit() as transaction:
        lease = transaction.claim_job_lease(
            job_id, WORKER, timedelta(minutes=5), agent=AGENT
        )
        transaction.commit()
    with lease:
        job = job_by_id(production, job_id)

        output.write_bytes(b"720p proxy essence\n")  # the actual work

        # Stage the output, the activity, and the completion in one transaction;
        # never commit the representation or activity separately.
        with production.edit() as transaction:
            proxy_id = transaction.add_representation(
                job.output_asset_id, job.output_representation_kind, output
            )
            activity_id = transaction.create_activity(
                ActivitySpec(
                    job.kind,
                    inputs=tuple(item for item in job.inputs),
                    outputs=(proxy_id,),
                    tool=WORKER,
                    agent=AGENT,
                )
            )
            # Copy the job parameters so the activity can be reproduced.
            for parameter in production.metadata[JobRef(job_id)]:
                transaction.add_metadata(
                    ActivityRef(activity_id), parameter.property, parameter.value
                )
            transaction.complete_job_lease(lease, proxy_id, activity_id)
            transaction.commit()

Rust

fn complete_proxy_job(
    production: &mut SqliteProduction,
    job_id: JobId,
    output_path: &Path,
) -> Result<(RepresentationId, ActivityId)> {
    let tool = ToolIdentity::new("Example Worker", Some("1.0".to_owned()), None)?;
    let job = production.job(job_id)?;
    let claim = {
        let mut transaction = production.begin_transaction()?;
        let claim = transaction.claim_job_lease(job_id, &tool, None, Duration::from_secs(60))?;
        transaction.commit()?;
        claim
    };

    // ... the worker writes the proxy file to output_path here ...
    let output = prepare_representation(
        job.requested_output().asset_id(),
        RepresentationKind::Proxy,
        output_path,
    )?;
    let activity = Activity::new(
        ActivityId::new(),
        ActivityKind::new(GENERATE_PROXY_JOB_KIND)?,
        job.inputs()
            .iter()
            .map(|input| ActivityInput::new(*input, None))
            .collect(),
        vec![ActivityOutput::new(output.representation().id(), None)],
    )?
    .with_tool(tool);

    let parameters = production.metadata(ObjectRef::Job(job_id))?;
    let mut transaction = production.begin_transaction()?;
    // Output, activity, and the succeeded state commit together or not at all.
    transaction.complete_job_lease(&claim, &output, &activity)?;
    // Keep the parameters on the activity so the artifact can be regenerated.
    for parameter in &parameters {
        transaction.add_metadata_value(
            ObjectRef::Activity(activity.id()),
            parameter.property(),
            parameter.value(),
        )?;
    }
    transaction.commit()?;
    Ok((output.representation().id(), activity.id()))
}

CLI

postproject --json job claim jobs.pproj "$JOB_ID" \
  --tool-name "Proxy Worker" --tool-version 1.0 \
  --lease 5m --lease-token-file completion.token
printf 'proxy media' > proxies/A001_proxy.mov
# Adds the output representation and its producing activity, and marks the
# job succeeded, in one transaction.
COMPLETED=$(postproject --json job complete jobs.pproj "$JOB_ID" \
  proxies/A001_proxy.mov --lease-token-file completion.token)
jq '{state, completion_representation_id, completion_activity_id, commit_receipt}' <<<"$COMPLETED"

Claiming from a read-derived edit rejects changed input fingerprints or complete dependency sets. Completion rechecks that knowledge against the claim’s revision or the edit’s older base. Renewal does not refresh the input decision. Changed knowledge returns a structured conflict and publishes no output or activity, including after reopening or importing the token. Re-read and perform fresh work under a new claim; changing the publication base alone cannot validate earlier work.

Completion validates that the activity consumes the job inputs and produces the staged output requested by the job. It does not copy the job’s parameters: record on the activity the parameters the worker actually used, so the output can be reproduced and a regeneration plan carries them. Do not commit the representation or activity in an earlier transaction: only complete gives the all-or-nothing guarantee. Library callers can stage any representation structure — single resource, image sequence, ordered parts, or package — before completing; the CLI completion adapter creates a single-file output.

Staging an output representation fingerprints its files, so a new output needs no separate fingerprint step. A worker that instead rewrites existing media in place must observe that content in the transaction:

a proxy regenerated at its old path;
a source replaced by a conform.

See recording a new fingerprint observation. Observation records the new resource value and recomputes every representation that uses the resource, so artifact evaluation reports the artifacts that depend on it as stale. Never write a representation fingerprint of your own devising. Only PostProject’s own values take part in staleness and resolution.

Some work happens before there is a production to record it in, for example proxies an editor renders before a project is first saved. Record such work later only when your application has its own evidence of what it was made from, and record its activity without a job, tool, or parameters you did not observe. Artifact evaluation can still report the output as current, and the reproducibility report says, correctly, that it cannot be regenerated.

Learn when work finishes¶

A requester does not need to poll the job list. Every job transition is a semantic revision event, so wait on a revision waiter and fetch the revisions filtered to JobSucceeded, JobFailed, and JobCancelled events after your cursor. This observes a worker in another process — for example postproject job run — as soon as it commits.

Public operations¶

Operation	Rust	C	C++	Python	CLI
request	`request_job`	`pp_transaction_request_job`	`requestJob`	`request_job`	`job request`
list	`jobs`	`pp_production_jobs`	`jobs`	`jobs`	`job list`
read one	`job`	`pp_production_job`	`job`	`job`	`job show`
claim	`claim_job_lease`	`pp_transaction_claim_job_lease`	`claimJobLease`	`claim_job_lease`	`job claim`
renew	`renew_job_lease`	`pp_transaction_renew_job_lease`	`renewJobLease`	`renew_job_lease`	`job renew`
release	`release_job_lease`	`pp_transaction_release_job_lease`	`releaseJobLease`	`release_job_lease`	`job release`
complete	`complete_job_lease`	`pp_transaction_complete_job_lease`	`completeJobLease`	`complete_job_lease`	`job complete`
fail	`fail_job_lease`	`pp_transaction_fail_job_lease`	`failJobLease`	`fail_job_lease`	`job fail`
cancel	`cancel_job`	`pp_transaction_cancel_job`	`cancelJob`	`cancel_job`	`job cancel`
plan regeneration	`plan_regeneration`	`pp_production_plan_regeneration`	`planRegeneration`	`plan_regeneration`	`job plan`

The C ABI returns owned job and regeneration-plan sets; reading one job returns a one-element job set. A job page’s optional cursor borrows the job set and must be copied before release. Release every returned set with its matching release function. The C++ and Python bindings copy result values and cursors into their native immutable types. The CLI accepts job list --state, --kind, --limit, and --cursor; JSON output is a page object with items and next_cursor.

The opt-in reference local executor builds on this protocol for proxy and thumbnail jobs. Its subprocess adapter is available in Rust and its complete runner is available as job run; C, C++, and Python workers use the protocol above rather than an executor wrapper.

Plan and explicitly enqueue regeneration¶

plan_regeneration accepts artifact representation IDs and returns proposals derived from recorded producers. Each proposal includes a fresh requested job and parameter metadata already retargeted to that job ID. The inputs and parameters come from the producing activity. When that activity completed a job, the proposal repeats the job’s kind and target root, so the work reaches the same workers and destination. Planning requires one unambiguous producer for each artifact (ADR 0023).

The proposal is not durable. To enqueue it, open a transaction, request the proposed job, copy its parameter assertions to the job target, and commit. This separation lets a host review, prioritize, modify, or discard proposed work and prevents a read from creating background work.

static pp_error_code_t enqueue_regeneration(pp_production_t *production,
                                            const pp_representation_id_t *artifact_id,
                                            pp_job_id_t *out_job_id,
                                            pp_error_t **error) {
  pp_regeneration_plan_set_t *plans = NULL;
  pp_job_set_t *planned = NULL;
  pp_metadata_set_t *parameters = NULL;
  pp_transaction_t *transaction = NULL;
  pp_representation_id_t planned_artifact;
  pp_job_t job;
  pp_representation_id_t inputs[16];
  memset(&job, 0, sizeof job);

  /* Planning is read-only; nothing is enqueued until you commit. */
  pp_error_code_t status = pp_production_plan_regeneration(
      production, artifact_id, 1, &plans, error);
  if (status == PP_OK && pp_regeneration_plan_set_count(plans) != 1) {
    status = PP_ERROR_NOT_FOUND;
  }
  if (status == PP_OK) {
    status = pp_regeneration_plan_set_get(plans, 0, &planned_artifact, &planned,
                                          &parameters, error);
  }
  if (status == PP_OK) {
    status = pp_job_set_get(planned, 0, &job, error);
  }
  if (status == PP_OK && job.input_count > 16) {
    status = PP_ERROR_UNSUPPORTED;
  }
  for (uint64_t i = 0; status == PP_OK && i < job.input_count; ++i) {
    status = pp_job_set_get_input(planned, 0, i, &inputs[i], error);
  }

  /* Enqueue the proposal explicitly, copying its parameters to the job. */
  if (status == PP_OK) {
    status = pp_production_begin_edit(production, NULL, NULL, &transaction, error);
  }
  if (status == PP_OK) {
    status = pp_transaction_request_job(
        transaction, job.kind, inputs, job.input_count, job.output_asset_id,
        job.output_representation_kind, job.target_root, out_job_id, error);
  }
  for (uint64_t i = 0; status == PP_OK && i < pp_metadata_set_count(parameters);
       ++i) {
    pp_object_ref_t target;
    status = pp_object_ref_from_job(*out_job_id, &target, error);
    if (status != PP_OK) break;
    pp_object_ref_t owner;
    const char *vocabulary, *property;
    const pp_metadata_value_t *value;
    pp_metadata_input_t *input = NULL;
    status = pp_metadata_set_get(parameters, i, &owner, &vocabulary, &property,
                                 &value, error);
    if (status == PP_OK) {
      status = metadata_input_from_value(value, &input, error);
    }
    if (status == PP_OK) {
      status = pp_transaction_add_metadata_value(
          transaction, &target, vocabulary, property, input, error);
    }
    pp_metadata_input_release(input);
  }
  if (status == PP_OK) {
    pp_commit_receipt_t commit_receipt;
    status = pp_transaction_commit(transaction, &commit_receipt, error);
  }

  pp_transaction_release(transaction);
  pp_metadata_set_release(parameters);
  pp_job_set_release(planned);
  pp_regeneration_plan_set_release(plans);
  return status;
}

C++

std::vector<postproject::JobId>
regenerate(postproject::Production &production,
           const postproject::RepresentationId &artifact_id) {
  // Planning is read-only: it derives requests from the producing activity.
  const auto plans = production.planRegeneration({artifact_id}).value();

  std::vector<postproject::JobId> job_ids;
  auto transaction = production.edit().value();
  for (const auto &plan : plans) {
    const postproject::JobRequest request{
        plan.job.kind, plan.job.inputs, plan.job.output_asset_id,
        plan.job.output_representation_kind, plan.job.target_root};
    const auto job_id = transaction.requestJob(request).value();
    for (const auto &parameter : plan.parameters) {
      transaction.addMetadataValue(postproject::ObjectRef::job(job_id),
                                   parameter.vocabulary, parameter.property,
                                   parameter.value).value();
    }
    job_ids.push_back(job_id);
  }
  transaction.commit().value();
  return job_ids;
}

Python

def plan_and_enqueue(
    production: Production, artifact_id: RepresentationId
) -> tuple[RegenerationJobPlan, JobId]:
    # Planning is read-only: it proposes a job but never enqueues one.
    (plan,) = production.plan_regeneration([artifact_id])
    proposed = plan.job
    print(f"regenerate {artifact_id} with {proposed.kind}")

    # Enqueue explicitly. The proposal repeats the kind and target root of the
    # job that produced the artifact.
    with production.edit() as transaction:
        job_id = transaction.request_job(
            JobRequest(
                proposed.kind,
                proposed.inputs,
                proposed.output_asset_id,
                proposed.output_representation_kind,
                proposed.target_root,
            )
        )
        # The planned parameters target the proposal; retarget them.
        for parameter in plan.parameters:
            transaction.add_metadata(
                JobRef(job_id), parameter.property, parameter.value
            )
        transaction.commit()
    return plan, job_id

Rust

fn regenerate(production: &mut SqliteProduction, artifact_id: RepresentationId) -> Result<JobId> {
    // Planning is read-only: nothing is enqueued until the caller requests it.
    let plans = production.plan_regeneration(&[artifact_id])?;
    let [plan] = plans.as_slice() else {
        panic!("one plan per artifact");
    };
    println!(
        "regenerate {} with {} ({} parameters)",
        plan.artifact_representation_id(),
        plan.job().kind().as_str(),
        plan.parameters().len()
    );

    let mut transaction = production.begin_transaction()?;
    transaction.request_job(plan.job())?;
    for parameter in plan.parameters() {
        transaction.add_metadata_value(
            ObjectRef::Job(plan.job().id()),
            parameter.property(),
            parameter.value(),
        )?;
    }
    transaction.commit()?;
    Ok(plan.job().id())
}

CLI

# Planning only reads recorded knowledge; nothing is enqueued until the
# planned request is submitted explicitly.
postproject --json job plan jobs.pproj --artifact "$PROXY_ID" > plan.json
jq '.[] | {artifact_representation_id, job: (.job | {kind, inputs, output_kind}),
  parameters}' plan.json
PLANNED=$(jq -c '.[0].job' plan.json)
INPUTS=()
for INPUT in $(jq -r '.inputs[]' <<<"$PLANNED"); do INPUTS+=(--input "$INPUT"); done
REGENERATION_ID=$(postproject --json job request jobs.pproj \
  "$(jq -r .kind <<<"$PLANNED")" "$(jq -r .output_asset_id <<<"$PLANNED")" \
  "$(jq -r .output_kind <<<"$PLANNED")" "${INPUTS[@]}" | jq -r .id)

See jobs and production work for the domain model and revision feed for cross-process change discovery.

Jobs and workers¶

Request work¶

Claim, renew, and release¶

Complete a job¶

Fail or cancel a job¶

Learn when work finishes¶

Public operations¶

Plan and explicitly enqueue regeneration¶