Exchange portable production knowledge

The development SDK can export a consistent checkpoint, create a passive mirror and catch it up with complete committed records. These operations are experimental; the released 0.7 SDK does not provide them. One authority owns the history. A mirror supports ordinary reads and local media resolution with the recipient’s root mappings; it rejects edits and worker authority.

Export a checkpoint and import it into a new destination. Validation uses the same complete domain checks in temporary private storage. Existing destinations and incomplete streams reject without publishing a partial mirror.

pp_exchange_position_t saved, validated;
pp_exchange_state_t mirror_state;
CHECK(pp_production_export_checkpoint(production, checkpoint, &saved, &error));
CHECK(pp_checkpoint_validate_file(checkpoint, NULL, &validated, &error));
if (memcmp(saved.digest, validated.digest, sizeof saved.digest)) goto cleanup;
CHECK(pp_checkpoint_import_file(checkpoint, mirror_path, NULL, &mirror, &error));
CHECK(pp_production_exchange_state(mirror, &mirror_state, &error));
if (mirror_state.role != PP_EXCHANGE_PASSIVE_MIRROR || !mirror_state.has_head || memcmp(mirror_state.head.digest, saved.digest, sizeof saved.digest)) goto cleanup;

const std::filesystem::path checkpoint = path + ".checkpoint";
const std::filesystem::path mirror_path = path + ".mirror.pproj";
POSTPROJECT_TRY_ASSIGN(auto saved, production.exportCheckpoint(checkpoint));
POSTPROJECT_TRY_ASSIGN(auto validated, Production::validateCheckpoint(checkpoint));
POSTPROJECT_TRY_ASSIGN(auto mirror, Production::importCheckpoint(checkpoint, mirror_path));
POSTPROJECT_TRY_ASSIGN(auto state, mirror.exchangeState());
if (saved.digest != validated.digest || state.role != ExchangeRole::passive_mirror ||
    !state.head || state.head->digest != saved.digest || state.protocol_version != 1 ||
    state.supported_features != UINT64_C(511))
  return Error(ErrorCode::internal, "checkpoint source boundary");

checkpoint = work / "checkpoint.ppxc"
saved = production.export_checkpoint(checkpoint)
assert Production.validate_checkpoint(checkpoint) == saved
mirror_path = work / "mirror.pproj"
with Production.import_checkpoint(checkpoint, mirror_path) as mirror:
    state = mirror.exchange_state
    assert state.role is ExchangeRole.PASSIVE_MIRROR
    assert state.head == saved

let mut output = File::create_new(&checkpoint_path)?;
let checkpoint = exchange_file::write_checkpoint(&authority, &mut output)?;
output.sync_all()?;
drop(output);
let mut mirror = exchange_file::import_checkpoint(
    &mirror_path,
    File::open(&checkpoint_path)?,
    FileLimits::default(),
    CheckpointLimits::default(),
)?;
assert_eq!(mirror.exchange_head()?, checkpoint.head());

postproject exchange position authority.pproj cursor.json
postproject exchange checkpoint export authority.pproj knowledge.ppxc
postproject exchange checkpoint validate knowledge.ppxc
postproject exchange checkpoint import knowledge.ppxc mirror.pproj

A proposal has stable client/request identities and ordered semantic commands. Keep those identities and prepared new-object IDs when retrying. Recover a lost reply by lookup or by submitting the same intent again. The returned receipt identifies that request’s own revision; an accepted no-op has no new revision. The CLI example submits a no-op; the Rust and C examples append an exact integer. Changed intent needs a new request. Destructive operations need a scoped decision base; a stale conflict requires a fresh read and a new decision.

C, C++ and Python prepare immutable public intent using typed transaction staging calls. Successful preparation closes the edit without writing a domain revision. Failed preparation leaves it open. Keep the returned proposal for submission and retries; pending native claims close during preparation.

pp_decision_base_t base;
pp_exchange_state_t state;
pp_object_ref_t target;
pp_media_root_id_t root;
const uint8_t *bytes;
uint64_t length;
/* Application-generated identities are retained for this complete intent. */
pp_client_id_t client = {{1}};
pp_request_id_t request = {{2}};
CHECK(pp_production_exchange_state(production, &state, &error));
if (state.protocol_version != 1 || state.role != PP_EXCHANGE_AUTHORITY || !state.has_head || state.floor.base.decision.revision_sequence != 0) goto cleanup;
CHECK(pp_production_read_session(production, &view, &error));
CHECK(pp_read_session_decision_base(view, &base, &error));
CHECK(pp_production_begin_edit(production, &base, NULL, &edit, &error));
target.kind = PP_OBJECT_PRODUCTION;
memcpy(target.id.bytes, base.production_id.bytes, sizeof target.id.bytes);
CHECK(pp_metadata_input_create_u64(UINT64_MAX, &value, &error));
CHECK(pp_transaction_add_metadata_value(edit, &target, "example:Exact", "count", value, &error));
CHECK(pp_transaction_add_media_root(edit, "rushes", NULL, -5, &root, &error));
CHECK(pp_transaction_prepare_proposal(edit, client, request, &state.observed_base, &proposal, &error));
CHECK(pp_proposal_document(proposal, &bytes, &length, &error));
if (length == 0 || bytes[0] != '{') goto cleanup;
/* bytes are borrowed public transport, with exact integers and prepared IDs. */

POSTPROJECT_TRY_ASSIGN(auto view, production.readSession());
POSTPROJECT_TRY_ASSIGN(auto base, view.decisionBase());
POSTPROJECT_TRY_ASSIGN(auto edit, production.edit(base));
POSTPROJECT_TRY(edit.addMetadataValue(ObjectRef::production(base.production_id),
    "example:Exact", "count", MetadataValue::unsignedInteger(UINT64_MAX)));
const ClientId client(Uuid::Bytes{1});
const RequestId request(Uuid::Bytes{2});
POSTPROJECT_TRY_ASSIGN(auto proposal, edit.prepareProposal(client, request, state.observed_base));
POSTPROJECT_TRY_ASSIGN(auto identity, proposal.identity());
POSTPROJECT_TRY_ASSIGN(auto bytes, proposal.document());
POSTPROJECT_TRY_ASSIGN(auto transported, Proposal::parse(bytes));
POSTPROJECT_TRY_ASSIGN(auto copied, transported.identity());
POSTPROJECT_TRY_ASSIGN(auto head, production.latestRevision());
if (head || copied.production_id != base.production_id ||
    identity.history_id != copied.history_id || copied.request_id != request ||
    copied.client_id != client || edit.commit().has_value())
  return Error(ErrorCode::internal, "prepared intent identity or lifecycle");

client, request = ClientId(uuid4()), RequestId(uuid4())
with production.read_session() as view, view.edit() as edit:
    edit.add_metadata(
        ProductionRef(production.id),
        MetadataProperty("example:Exact", "count"),
        MetadataU64(2**64 - 1),
    )
    edit.add_media_root("rushes", priority=-5)
    proposal = edit.prepare_proposal(client, request)

Construct checked Command values directly, as shown below.

Submit a portable proposal file, as shown below.

pp_request_identity_t identity;
pp_commit_receipt_t receipt;
uint32_t status;
CHECK(pp_proposal_identity(proposal, &identity, &error));
CHECK(pp_proposal_parse(bytes, length, &transported, &error));
CHECK(pp_production_submit_proposal(production, transported, NULL, 0, NULL, 0, &submission, &error));
CHECK(pp_submission_outcome(submission, &outcome, &error));
if (pp_submission_lease_count(submission) != 0) goto cleanup;
pp_job_lease_t *absent = NULL;
if (pp_submission_take_lease(submission, 0, &absent, NULL) != PP_ERROR_INVALID_ARGUMENT || absent != NULL) goto cleanup;
CHECK(pp_outcome_status(outcome, &status, &error));
if (status != PP_OUTCOME_ACCEPTED) goto cleanup;
CHECK(pp_outcome_receipt(outcome, &receipt, &error));
if (receipt.outcome != PP_COMMIT_REVISION_CREATED || receipt.revision_sequence != 1) goto cleanup;
CHECK(pp_outcome_rejection(outcome, &rejection, &error));
if (rejection != NULL) goto cleanup;
pp_submission_release(submission); submission = NULL;
/* Simulate response loss: reopen and recover the original scoped request. */
pp_production_release(production); production = NULL;
CHECK(pp_production_open(path, &production, &error));
CHECK(pp_production_submission_outcome(production, &identity, &recovered, &error));
const uint8_t *original_bytes, *recovered_bytes;
uint64_t original_length, recovered_length;
CHECK(pp_outcome_document(outcome, &original_bytes, &original_length, &error));
CHECK(pp_outcome_document(recovered, &recovered_bytes, &recovered_length, &error));
if (original_length != recovered_length || memcmp(original_bytes, recovered_bytes, (size_t)original_length)) goto cleanup;
CHECK(pp_production_submit_proposal(production, proposal, NULL, 0, NULL, 0, &duplicate, &error));
if (pp_submission_lease_count(duplicate) != 0) goto cleanup;
pp_outcome_release(outcome); outcome = NULL;
CHECK(pp_submission_outcome(duplicate, &outcome, &error));
CHECK(pp_outcome_receipt(outcome, &receipt, &error));
if (receipt.revision_sequence != 1) goto cleanup;
pp_outcome_release(recovered); recovered = NULL;

POSTPROJECT_TRY_ASSIGN(auto submission, production.submitProposal(transported));
const auto *receipt = std::get_if<CommitReceipt>(&submission.outcome.status);
if (!receipt || !receipt->revision || receipt->revision->sequence != 1 ||
    !submission.leases.empty() || !submission.outcome.jobs.empty())
  return Error(ErrorCode::internal, "own submitted receipt");
POSTPROJECT_TRY_ASSIGN(production, Production::open(path));
POSTPROJECT_TRY_ASSIGN(auto recovered, production.submissionOutcome(identity));
POSTPROJECT_TRY_ASSIGN(auto duplicate, production.submitProposal(proposal));
if (recovered.document != submission.outcome.document ||
    duplicate.outcome.document != recovered.document || !duplicate.leases.empty())
  return Error(ErrorCode::internal, "lost reply and duplicate recovery");

with proposal, Proposal.parse(proposal.document) as transported:
    identity = proposal.identity
    assert transported.identity == identity
    assert identity.client_id == client and identity.request_id == request
    submitted = production.submit_proposal(transported)
    assert isinstance(submitted.outcome.status, CommitReceipt)
    assert submitted.outcome.status.revision is not None
    assert submitted.outcome.status.revision.sequence == 1
    assert not submitted.leases
    with Production.open(work / "authority.pproj") as reopened:
        recovered = reopened.submission_outcome(identity)
        duplicate = reopened.submit_proposal(proposal)
        assert recovered.document == submitted.outcome.document
        assert duplicate.outcome.document == recovered.document
        assert not duplicate.leases

let target = ObjectRef::Production(authority.production().id());
let property = MetadataProperty::new(
    VocabularyId::new("urn:studio:opaque")?,
    PropertyId::new("sample")?,
);
let proposal = Proposal::new(
    authority.exchange_scope()?,
    ClientId::new(),
    RequestId::new(),
    None,
    RevisionContext::default(),
    vec![Command::AppendMetadata {
        target,
        property: property.clone(),
        value: MetadataValue::u64((1 << 53) + 1),
    }],
    Extensions::default(),
)?;
let accepted = authority.submit_proposal(&proposal)?;
drop(authority);
let mut authority = SqliteProduction::open(&authority_path)?;
let recovered =
    authority.submission_outcome(proposal.scope(), proposal.client(), proposal.request())?;
assert_eq!(recovered.as_ref(), Some(&accepted));
assert_eq!(authority.submit_proposal(&proposal)?, accepted);

CLIENT=00000000-0000-4000-8000-000000000001
REQUEST=00000000-0000-4000-8000-000000000002
jq --arg client "$CLIENT" --arg request "$REQUEST" '{
  kind: "proposal", version: "1", required_features: ["metadata.v1"],
  production, history, client: $client, request: $request,
  base: null, origin: null, message: null, commands: [], extensions: {}
}' cursor.json > proposal.json
postproject --json exchange submit authority.pproj proposal.json > accepted.json
postproject --json exchange outcome authority.pproj \
  --history "$(jq -r .history cursor.json)" \
  --client "$CLIENT" --request "$REQUEST" > recovered.json
test "$(jq -c .outcome accepted.json)" = "$(jq -c .outcome recovered.json)"
test "$(jq -r .outcome.receipt.revision accepted.json)" = null

Python also accepts ordered command collections. Staging returns typed prepared IDs for later commands. Paths and leases are local inputs; prepared portable intent contains measured facts and public identities. After a staging failure, discard the builder before making a new decision.

Stage commands with the typed transaction calls shown above.

Stage commands with the typed transaction methods shown above.

with production.proposal_builder(
    ClientId(uuid4()),
    RequestId(uuid4()),
    base=state.observed_base,
) as builder:
    results = builder.stage([AddMediaRootCommand("discarded")])
    assert isinstance(results[0], PreparedMediaRoot)
    with builder.prepare() as discarded:
        assert discarded.identity.history_id == state.observed_base.history_id
    builder.close()
assert not production.media_roots  # Preparation/discard never commits.

Construct checked Command values directly.

Submit a portable proposal file.

Catch-up exports every complete record after a saved source position through one pinned head. Ordinary native transactions enter that history automatically. Positions retain production, history, revision, sequence and digest. Observation feed cursors and connection-local query cursors cannot substitute for them.

pp_exchange_position_t through, applied;
CHECK(pp_production_export_changes(production, &saved, changes, &through, &error));
CHECK(pp_production_apply_changes(mirror, changes, NULL, &applied, &error));
if (applied.base.decision.revision_sequence != 1 || memcmp(through.digest, applied.digest, sizeof applied.digest)) goto cleanup;
pp_production_release(mirror); mirror = NULL;
CHECK(pp_production_open(mirror_path, &mirror, &error));
CHECK(pp_production_apply_changes(mirror, changes, NULL, &applied, &error));
CHECK(pp_production_exchange_state(mirror, &mirror_state, &error));
if (mirror_state.head.base.decision.revision_sequence != 1 || memcmp(mirror_state.head.digest, through.digest, sizeof through.digest)) goto cleanup;
/* Ordinary read APIs see the authority's exact copied knowledge. */
pp_media_root_set_release(roots); roots = NULL;
CHECK(pp_production_media_roots(mirror, &roots, &error));
if (pp_media_root_set_count(roots) != 1) goto cleanup;
CHECK(pp_production_metadata(mirror, &target, &metadata, &error));
pp_object_ref_t copied_target;
const pp_metadata_value_t *copied;
const char *vocabulary, *property;
uint64_t exact;
CHECK(pp_metadata_set_get(metadata, 0, &copied_target, &vocabulary, &property, &copied, &error));
CHECK(pp_metadata_value_get_u64(copied, &exact, &error));
if (copied_target.kind != PP_OBJECT_PRODUCTION || memcmp(copied_target.id.bytes, target.id.bytes, sizeof target.id.bytes) || strcmp(vocabulary, "example:Exact") || strcmp(property, "count")) goto cleanup;
pp_metadata_set_release(metadata); metadata = NULL;
if (exact != UINT64_MAX) goto cleanup;

const std::filesystem::path changes = path + ".changes";
POSTPROJECT_TRY_ASSIGN(auto through, production.exportChanges(saved, changes));
POSTPROJECT_TRY_ASSIGN(auto applied, mirror.applyChanges(changes));
if (through.digest != applied.digest || !applied.base.decision.revision ||
    applied.base.decision.revision->sequence != 1)
  return Error(ErrorCode::internal, "complete catch-up boundary");
POSTPROJECT_TRY_ASSIGN(mirror, Production::open(path + ".mirror.pproj"));
POSTPROJECT_TRY_ASSIGN(applied, mirror.applyChanges(changes));
POSTPROJECT_TRY_ASSIGN(state, mirror.exchangeState());
if (!state.head || applied.digest != state.head->digest)
  return Error(ErrorCode::internal, "restart and duplicate catch-up");
POSTPROJECT_TRY_ASSIGN(auto exact, mirror.queryMetadata("example:Exact", "count", 10));
if (exact.items.size() != 1 ||
    exact.items[0].value.getIf<MetadataUnsignedInteger>()->value != UINT64_MAX)
  return Error(ErrorCode::internal, "ordinary exact metadata read");

changes = work / "changes.ppxd"
through = production.export_changes(saved, changes)
with Production.open(mirror_path) as mirror:
    assert mirror.apply_changes(changes) == through
with Production.open(mirror_path) as mirror:
    assert mirror.apply_changes(changes) == through
    assert mirror.exchange_state.head == through
    assert mirror.metadata[ProductionRef(mirror.id)][0].value == MetadataU64(
        2**64 - 1
    )
    assert len(mirror.media_roots) == 1

let mut output = File::create_new(&changes_path)?;
let through = exchange_file::write_changes(&authority, checkpoint.head(), &mut output)?;
output.sync_all()?;
drop(output);
assert_eq!(
    exchange_file::apply_changes(
        &mut mirror,
        File::open(&changes_path)?,
        FileLimits::default(),
        ReplayLimits::default()
    )?,
    through
);
assert_eq!(
    mirror.metadata_values(target, &property)?,
    authority.metadata_values(target, &property)?
);
assert!(mirror.begin_transaction().is_err());

# Ordinary native writes automatically enter the complete source history.
postproject --json media add authority.pproj rushes/A001.mov > asset.json
postproject exchange changes export authority.pproj changes.ppxd --from cursor.json
postproject --json exchange changes apply mirror.pproj changes.ppxd > applied.json
postproject --json exchange inspect mirror.pproj > mirror.json
test "$(jq -c .head applied.json)" = "$(jq -c .head mirror.json)"
test "$(jq -r .role mirror.json)" = passive_mirror

Each applied record is atomic. If later input fails or the process stops, earlier complete records remain durable. Reopen, inspect the mirror head and retry the same stream; verified duplicates create no revision. A missing predecessor or pre-floor position reports history_gap and requires a checkpoint. Unsupported required features reject; old histories are never guessed or merged.

Incomplete early development history requires an explicit authority recovery: save exchange history decision authority.pproj decision.json, then run exchange history resynchronize authority.pproj --base decision.json. The operation checks a fresh scoped head, preserves earlier bytes, revisions and private outcomes, and establishes a new checkpoint floor only when necessary. Complete histories keep their floor. Export a new checkpoint for affected mirrors; recovery cannot promote a mirror or reconstruct missing bytes.

Files carry bounded chunks, not media bytes. Defaults allow 1 GiB encoded input; checkpoint import also allows 2 GiB private database/journal space and 10 million frames/traversal operations. Raise receiver budgets explicitly for larger productions. Rust writers return their final seal after streaming; publish output only after success and durable flush. CLI exports publish new files exclusively. Long pinned exports can retain SQLite WAL pages.

Job observations retain attribution, lifecycle and publication evidence without usable credentials. Worker submissions use private lease handles or token files; see CLI output for explicit CLI delivery and recovery. Duplicate lookup never reissues ownership. Lost credential delivery needs expiry or coordinator cancellation.

Hashes detect inconsistency relative to the saved source anchor; they do not authenticate a source or prevent two copied authorities from diverging. No automatic pruning, media transfer, network service, editable replica or authority restore is provided. A checkpoint cannot restore private outcomes, clocks or worker capabilities.