Implement crash-safe streaming worker runtime

Cargo.lock

@@ -3522,6 +3522,42 @@ dependencies = [
 [[package]]
 name = "remora-worker"
 version = "0.0.0"
+dependencies = [
+ "actix-tls",
+ "actix-web",
+ "base64",
+ "bytes",
+ "futures-util",
+ "getrandom 0.3.4",
+ "http 1.4.2",
+ "rcgen",
+ "remora-admission",
+ "remora-api",
+ "remora-attempt",
+ "remora-contracts",
+ "remora-coordination",
+ "remora-credentials",
+ "remora-domain",
+ "remora-durable",
+ "remora-hf-source",
+ "remora-http-source",
+ "remora-identity",
+ "remora-object-source",
+ "remora-outbound",
+ "remora-result",
+ "remora-stream",
+ "remora-target",
+ "reqwest 0.13.4",
+ "rustls",
+ "serde",
+ "serde_json",
+ "serde_json_canonicalizer",
+ "sha2 0.11.0",
+ "sqlx",
+ "thiserror",
+ "time",
+ "tokio",
+]
 
 [[package]]
 name = "reqsign-aws-v4"

Dockerfile.test

@@ -285,6 +285,35 @@ COPY crates crates
 COPY --from=kind-worker-api-verified /tmp/remora-api-tests.tar.zst /tests/remora-api-tests.tar.zst
 CMD ["bash", "-ec", "test -s /run/remora-fixture/profile.env && test -n \"$DATABASE_URL\" && test -n \"$REMORA_COORDINATION_URL\" && export REMORA_WORKER_API_REQUIRE=1 && cargo-nextest nextest run --archive-file /tests/remora-api-tests.tar.zst --workspace-remap /workspace --test-threads=1 --no-fail-fast"]
 
+FROM rust-tools AS kind-worker-build
+WORKDIR /workspace
+COPY Cargo.toml Cargo.lock rust-toolchain.toml ./
+COPY .config .config
+COPY contracts contracts
+COPY crates crates
+COPY docs/api docs/api
+RUN cargo nextest archive \
+    --package remora-worker \
+    --build-jobs 4 \
+    --locked \
+    --archive-file /tmp/remora-worker-tests.tar.zst
+
+FROM kind-worker-build AS kind-worker-verified
+RUN cargo clippy \
+    --package remora-worker \
+    --all-targets \
+    --all-features \
+    --locked \
+    -- -D warnings \
+    && cargo test --package remora-worker --doc --locked
+
+FROM rust-tools AS kind-worker-runner
+WORKDIR /workspace
+COPY Cargo.toml ./
+COPY crates crates
+COPY --from=kind-worker-verified /tmp/remora-worker-tests.tar.zst /tests/remora-worker-tests.tar.zst
+CMD ["bash", "-ec", "test -s /run/remora-fixture/profile.env && test -n \"$DATABASE_URL\" && test -n \"$REMORA_COORDINATION_URL\" && test -n \"$REMORA_TARGET_ENDPOINT\" && test -n \"$REMORA_VAULT_ADDR\" && case $REMORA_COORDINATION_PROVIDER in redis|dragonfly) ;; *) exit 1 ;; esac && case $REMORA_TARGET_PROVIDER in garage|versitygw) ;; *) exit 1 ;; esac && export REMORA_WORKER_REQUIRE=1 && cargo-nextest nextest run --archive-file /tests/remora-worker-tests.tar.zst --workspace-remap /workspace --test-threads=1 --no-fail-fast"]
+
 FROM rust-tools AS kind-http-source-build
 WORKDIR /workspace
 COPY Cargo.toml Cargo.lock rust-toolchain.toml ./

Makefile

@@ -1,4 +1,4 @@
-.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-admission check-attempt check-attempt-mutations check-recovery check-worker-api check-stream check-target check-result check-outbound check-http-source check-object-source check-hf-source check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
+.PHONY: check-dependencies check-fixture-lib check-kind-sample check-durable check-coordination check-identity check-credentials check-admission check-attempt check-attempt-mutations check-recovery check-worker-api check-worker check-stream check-target check-result check-outbound check-http-source check-object-source check-hf-source check-fast check-full check-fast-host check-full-host rust-check web-install web-check web-test-fast web-test-full
 
 CONTAINER_ENGINE ?= nerdctl
 CONTAINER_RUN_FLAGS ?= --memory=16g --memory-swap=24g --cpus=4
@@ -83,6 +83,9 @@ check-recovery:
 check-worker-api:
 	./tests/kind/run-worker-api-job.sh
 
+check-worker:
+	./tests/kind/run-worker-job.sh
+
 check-stream:
 	cargo fmt --all --check
 	cargo clippy --package remora-stream --all-targets --all-features -- -D warnings

crates/remora-api/src/worker.rs

@@ -8,22 +8,25 @@ use actix_web::{
 };
 use remora_attempt::{
     ActivationMutationOutcome, AttemptAuthority, AttemptError, AttemptMutationOutcome,
-    AttemptService, CheckpointProjection, CheckpointRecord, ClaimOutcome, DurableDescriptor,
+    AttemptService, CheckpointMultipart, CheckpointProjection, CheckpointRecord, ClaimOutcome,
-    LeaseProjection, MutationRejection, OutcomePublication, ProgressRecord, RenewalMutationOutcome,
+    DurableDescriptor, LeaseProjection, MutationRejection, OutcomePublication, ProgressRecord,
+    RenewalMutationOutcome, StagingMultipart, StagingObject, StagingWriteIntent,
 };
 use remora_contracts::{
     ActivationResponseDto, ApiErrorCodeDto, ApiErrorDto, AttemptActivateRequestDto,
     AttemptAuthorityDto, AttemptCheckpointRequestDto, AttemptOutcomeDto, AttemptOutcomeRequestDto,
     AttemptProgressRequestDto, AttemptRenewRequestDto, AttemptResolvedSourceRequestDto,
-    CheckpointDescriptorKindDto, CheckpointDescriptorRefDto, ClaimRequestDto, ClaimResponseDto,
+    AttemptStagingMultipartRequestDto, AttemptStagingObjectRequestDto,
-    CredentialGrantVersionDto, DateTimeDto, DurableDescriptorKindDto, DurableDescriptorRefDto,
+    AttemptStagingWriteIntentRequestDto, AttemptStopRequestDto, CheckpointDescriptorKindDto,
-    ExecutionEnvelopeDto, JobPhaseProjectionDto, LeaseProjectionDto, NullableDto,
+    CheckpointDescriptorRefDto, ClaimRequestDto, ClaimResponseDto, CredentialGrantVersionDto,
-    PositiveIntegerDto, ResolvedSourceDescriptorKindDto, ResolvedSourceDescriptorRefDto,
+    DateTimeDto, DurableDescriptorKindDto, DurableDescriptorRefDto, ExecutionEnvelopeDto,
-    SafeIntegerDto, SourceProjectionDto, StagingTargetDto, TargetDto,
+    JobPhaseProjectionDto, LeaseProjectionDto, NullableDto, PositiveIntegerDto,
+    ResolvedSourceDescriptorKindDto, ResolvedSourceDescriptorRefDto, SafeIntegerDto,
+    SourceProjectionDto, StagingTargetDto, TargetDto,
 };
 use remora_domain::OpaqueId;
 use remora_result::PublicationInput;
-use remora_target::{ContentSha256, ObjectIdentity, RESERVED_NAMESPACE_SEGMENT, StoredObject};
+use remora_target::{ContentSha256, ObjectIdentity, StoredObject, attempt_staging_prefix};
 use sha2::{Digest, Sha256};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 
@@ -96,6 +99,19 @@ pub fn configure_worker_api(config: &mut web::ServiceConfig) {
             .route("/jobs/{job_id}/claim", web::post().to(claim))
             .route("/attempts/{attempt_id}/renew", web::post().to(renew))
             .route("/attempts/{attempt_id}/activate", web::post().to(activate))
+            .route("/attempts/{attempt_id}/stop", web::post().to(stop))
+            .route(
+                "/attempts/{attempt_id}/staging/write-intent",
+                web::post().to(register_staging_write_intent),
+            )
+            .route(
+                "/attempts/{attempt_id}/staging/object",
+                web::post().to(register_staging_object),
+            )
+            .route(
+                "/attempts/{attempt_id}/staging/multipart",
+                web::post().to(register_staging_multipart),
+            )
             .route(
                 "/attempts/{attempt_id}/resolved-source",
                 web::post().to(pin_resolved_source),
@@ -115,6 +131,137 @@ pub fn configure_worker_api(config: &mut web::ServiceConfig) {
     );
 }
 
+async fn register_staging_object(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingObjectRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_object_operation(
+            &authority,
+            &request.operation_id,
+            StagingObject {
+                target_store: request.object.store.clone(),
+                bucket: request.object.bucket.clone(),
+                key: request.object.key.clone(),
+                version_id: request.object.version_id.clone(),
+            },
+            request.completed_upload_id.clone(),
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}
+
+async fn register_staging_write_intent(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingWriteIntentRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_write_intent_operation(
+            &authority,
+            &request.operation_id,
+            StagingWriteIntent {
+                target_store: request.intent.store.clone(),
+                bucket: request.intent.bucket.clone(),
+                key: request.intent.key.clone(),
+            },
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}
+
+async fn register_staging_multipart(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStagingMultipartRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    let accepted = state
+        .attempts
+        .register_staging_multipart_operation(
+            &authority,
+            &request.operation_id,
+            StagingMultipart {
+                target_store: request.multipart.store.clone(),
+                bucket: request.multipart.bucket.clone(),
+                key: request.multipart.key.clone(),
+                upload_id: request.multipart.upload_id.clone(),
+            },
+        )
+        .await
+        .map_err(WorkerApiError::from_attempt)?;
+    if accepted {
+        Ok(HttpResponse::NoContent().finish())
+    } else {
+        Err(WorkerApiError::precondition_failed())
+    }
+}
+
+async fn stop(
+    _: AuthenticatedWorker,
+    state: web::Data<WorkerApiState>,
+    path: web::Path<String>,
+    request: web::Json<AttemptStopRequestDto>,
+) -> Result<HttpResponse, WorkerApiError> {
+    let authority = request_authority(
+        &state,
+        path.into_inner(),
+        request.fence.get(),
+        &request.worker_incarnation,
+    )
+    .await?;
+    match state
+        .attempts
+        .cooperative_stop_operation(&authority, &request.operation_id)
+        .await
+        .map_err(WorkerApiError::from_attempt)?
+    {
+        remora_attempt::CooperativeStopOutcome::Stopped
+        | remora_attempt::CooperativeStopOutcome::StopNotRequired => {
+            Ok(HttpResponse::NoContent().finish())
+        }
+        remora_attempt::CooperativeStopOutcome::AuthorityMismatch => {
+            Err(WorkerApiError::precondition_failed())
+        }
+    }
+}
+
 async fn claim(
     _: AuthenticatedWorker,
     state: web::Data<WorkerApiState>,
@@ -245,6 +392,14 @@ async fn persist_checkpoint(
                     resolved_source_id: request.resolved_source_id.clone(),
                     validated_bytes: request.validated_bytes.get(),
                     validated_members: request.validated_members.get(),
+                    staging_multipart: request.staging_multipart.as_ref().map(|identity| {
+                        CheckpointMultipart {
+                            target_store: identity.store.clone(),
+                            bucket: identity.bucket.clone(),
+                            key: identity.key.clone(),
+                            upload_id: identity.upload_id.clone(),
+                        }
+                    }),
                 },
             )
             .await
@@ -387,6 +542,7 @@ async fn load_execution_envelope(
     let source: SourceProjectionDto =
         serde_json::from_value(row.source).map_err(|_| WorkerApiError::unavailable())?;
     let target_prefix = row.target_prefix.join("/");
+    let staging_prefix = attempt_staging_prefix(&target_prefix, &row.staging_origin_attempt_id);
     let resolved_source_ref = row.resolved_source.map(|descriptor| {
         NullableDto(Some(descriptor_ref(
             descriptor,
@@ -425,10 +581,7 @@ async fn load_execution_envelope(
         staging_target: StagingTargetDto {
             store: row.target_store,
             bucket: row.target_bucket,
-            prefix: format!(
+            prefix: staging_prefix,
-                "{RESERVED_NAMESPACE_SEGMENT}/a/{}/",
-                authority.attempt_id.as_str()
-            ),
         },
         execution_grant_ref: row.execution_grant_id,
         checkpoint_ref: checkpoint_ref.unwrap_or(NullableDto(None)),

crates/remora-api/tests/worker_api.rs

@@ -15,11 +15,13 @@ use remora_attempt::{AttemptConfig, AttemptService, RecoveryConfig, migrate as a
 use remora_contracts::{
     ApiErrorDto, AttemptActivateRequestDto, AttemptCheckpointRequestDto, AttemptFailureOutcomeDto,
     AttemptOutcomeDto, AttemptOutcomeRequestDto, AttemptProgressRequestDto, AttemptRenewRequestDto,
-    AttemptResolvedSourceRequestDto, AttemptRuntimeFailureDto, CheckpointDescriptorKindDto,
+    AttemptResolvedSourceRequestDto, AttemptRuntimeFailureDto, AttemptStagingMultipartRequestDto,
-    CheckpointDescriptorRefDto, ClaimRequestDto, ClaimResponseDto, HttpSourceSubmissionDto,
+    AttemptStagingObjectRequestDto, AttemptStagingWriteIntentRequestDto, AttemptStopRequestDto,
-    JobPhaseProjectionDto, JobSubmissionDto, PositiveIntegerDto, ProgressValueDto,
+    CheckpointDescriptorKindDto, CheckpointDescriptorRefDto, ClaimRequestDto, ClaimResponseDto,
-    ResolvedSourceDescriptorKindDto, ResolvedSourceDescriptorRefDto, ResultDescriptorDto,
+    HttpSourceSubmissionDto, JobPhaseProjectionDto, JobSubmissionDto, PositiveIntegerDto,
-    ResultKindDto, SafeIntegerDto, SourceSubmissionDto,
+    ProgressValueDto, ResolvedSourceDescriptorKindDto, ResolvedSourceDescriptorRefDto,
+    ResultDescriptorDto, ResultKindDto, SafeIntegerDto, SourceSubmissionDto,
+    StagingMultipartIdentityDto, StagingObjectIdentityDto, StagingWriteIntentDto,
 };
 use remora_coordination::{CoordinationStore, StoreConfig};
 use remora_credentials::migrate as credentials_migrate;
@@ -29,7 +31,7 @@ use remora_domain::{
 use remora_durable::{PoolConfig, connect, migrate as durable_migrate};
 use remora_identity::{DefaultTarget, Principal};
 use remora_result::migrate as result_migrate;
-use remora_target::attempt_manifest_key;
+use remora_target::attempt_manifest_key_in;
 use reqwest::{Certificate as ClientCertificate, Client, Identity};
 use rustls::{
     RootCertStore, ServerConfig,
@@ -153,14 +155,15 @@ impl Harness {
     async fn seal_result(&self, attempt_id: &OpaqueId) -> ResultDescriptorDto {
         let manifest_id = unique_id("manifest");
         let checksum = format!("sha256:{}", "11".repeat(32));
-        let key = attempt_manifest_key(attempt_id, &manifest_id);
+        let key = attempt_manifest_key_in("users/alice", attempt_id, &manifest_id);
         sqlx::query(
             "INSERT INTO result_inventories (
                 attempt_id, state, manifest_schema_version, manifest_id, manifest_store,
                 manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
-                manifest_size, member_count, total_bytes
+                manifest_size, member_count, total_bytes,
+                staging_origin_attempt_id, target_prefix
              ) VALUES ($1, 'sealed', 1, $2, 'primary', 'remora-results', $3,
-                       'version-1', $4, 128, 1, 4096)",
+                       'version-1', $4, 128, 1, 4096, $1, 'users/alice')",
         )
         .bind(attempt_id.as_str())
         .bind(manifest_id.as_str())
@@ -336,6 +339,92 @@ async fn worker_operations_are_authenticated_fenced_replayable_and_redacted() {
         assert_eq!(response.status(), 409);
     }
 
+    let staging_key = format!(
+        "{}o/{}",
+        claimed.execution.staging_target.prefix,
+        unique_id("object")
+    );
+    let staging_intent = AttemptStagingWriteIntentRequestDto {
+        operation_id: unique_id("operation"),
+        fence: claimed.authority.fence,
+        worker_incarnation: worker.clone(),
+        intent: StagingWriteIntentDto {
+            store: claimed.execution.target.store.clone(),
+            bucket: claimed.execution.target.bucket.clone(),
+            key: staging_key.clone(),
+        },
+    };
+    for _ in 0..2 {
+        assert_eq!(
+            client
+                .post(format!("{attempt_url}/staging/write-intent"))
+                .json(&staging_intent)
+                .send()
+                .await
+                .unwrap()
+                .status(),
+            204
+        );
+    }
+    let staging_multipart = AttemptStagingMultipartRequestDto {
+        operation_id: unique_id("operation"),
+        fence: claimed.authority.fence,
+        worker_incarnation: worker.clone(),
+        multipart: StagingMultipartIdentityDto {
+            store: claimed.execution.target.store.clone(),
+            bucket: claimed.execution.target.bucket.clone(),
+            key: staging_key.clone(),
+            upload_id: "upload-identity".into(),
+        },
+    };
+    assert_eq!(
+        client
+            .post(format!("{attempt_url}/staging/multipart"))
+            .json(&staging_multipart)
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        204
+    );
+    let staging_operation = unique_id("operation");
+    let staging_object = AttemptStagingObjectRequestDto {
+        operation_id: staging_operation.clone(),
+        fence: claimed.authority.fence,
+        worker_incarnation: worker.clone(),
+        object: StagingObjectIdentityDto {
+            store: claimed.execution.target.store.clone(),
+            bucket: claimed.execution.target.bucket.clone(),
+            key: staging_key.clone(),
+            version_id: None,
+        },
+        completed_upload_id: Some("upload-identity".into()),
+    };
+    for _ in 0..2 {
+        assert_eq!(
+            client
+                .post(format!("{attempt_url}/staging/object"))
+                .json(&staging_object)
+                .send()
+                .await
+                .unwrap()
+                .status(),
+            204
+        );
+    }
+    let mut changed_staging = staging_object;
+    changed_staging.object.key.push_str("-changed");
+    assert_eq!(
+        client
+            .post(format!("{attempt_url}/staging/object"))
+            .json(&changed_staging)
+            .send()
+            .await
+            .unwrap()
+            .status(),
+        409
+    );
+
     let resolved_id = unique_id("resolved");
     let resolved = AttemptResolvedSourceRequestDto {
         operation_id: unique_id("operation"),
@@ -375,6 +464,7 @@ async fn worker_operations_are_authenticated_fenced_replayable_and_redacted() {
         resolved_source_id: resolved_id,
         validated_bytes: safe(1024),
         validated_members: safe(1),
+        staging_multipart: None,
     };
     assert_eq!(
         client
@@ -470,7 +560,7 @@ async fn worker_operations_are_authenticated_fenced_replayable_and_redacted() {
         .json(&AttemptProgressRequestDto {
             operation_id: renewal_operation,
             fence: claimed.authority.fence,
-            worker_incarnation: worker,
+            worker_incarnation: worker.clone(),
             progress: ProgressValueDto {
                 completed_bytes: safe(2048),
                 total_bytes: Some(safe(4096)),
@@ -488,6 +578,23 @@ async fn worker_operations_are_authenticated_fenced_replayable_and_redacted() {
     );
     let error: ApiErrorDto = cross_kind.json().await.unwrap();
     assert_eq!(error.message, "precondition failed");
+    let stop = AttemptStopRequestDto {
+        operation_id: unique_id("operation"),
+        fence: claimed.authority.fence,
+        worker_incarnation: worker,
+    };
+    for _ in 0..2 {
+        assert_eq!(
+            client
+                .post(format!("{attempt_url}/stop"))
+                .json(&stop)
+                .send()
+                .await
+                .unwrap()
+                .status(),
+            204
+        );
+    }
     server.stop(true).await;
 }
 

crates/remora-attempt/migrations/0005_checkpoint_staging_ownership.sql

@@ -0,0 +1,39 @@
+ALTER TABLE attempts
+    ADD COLUMN staging_origin_attempt_id TEXT;
+
+UPDATE attempts
+SET staging_origin_attempt_id = attempt_id;
+
+ALTER TABLE attempts
+    ALTER COLUMN staging_origin_attempt_id SET NOT NULL,
+    ADD CONSTRAINT attempts_staging_origin_fk
+        FOREIGN KEY (staging_origin_attempt_id)
+        REFERENCES attempts (attempt_id)
+        DEFERRABLE INITIALLY DEFERRED,
+    ADD CONSTRAINT attempts_id_job_unique UNIQUE (attempt_id, job_id);
+
+ALTER TABLE jobs
+    ADD COLUMN checkpoint_owner_attempt_id TEXT;
+
+UPDATE jobs j
+SET checkpoint_owner_attempt_id = (
+    SELECT a.attempt_id
+    FROM attempts a
+    WHERE a.job_id = j.job_id
+    ORDER BY a.fence DESC
+    LIMIT 1
+)
+WHERE j.checkpoint IS NOT NULL;
+
+ALTER TABLE jobs
+    ADD CONSTRAINT jobs_checkpoint_owner_fk
+        FOREIGN KEY (checkpoint_owner_attempt_id, job_id)
+        REFERENCES attempts (attempt_id, job_id)
+        DEFERRABLE INITIALLY DEFERRED,
+    ADD CONSTRAINT jobs_checkpoint_owner_check CHECK (
+        (checkpoint IS NULL) = (checkpoint_owner_attempt_id IS NULL)
+    );
+
+CREATE INDEX jobs_checkpoint_owner_idx
+    ON jobs (checkpoint_owner_attempt_id)
+    WHERE checkpoint_owner_attempt_id IS NOT NULL;

crates/remora-attempt/migrations/0006_staging_write_intents.sql

@@ -0,0 +1,9 @@
+ALTER TABLE attempt_staging_cleanup
+    DROP CONSTRAINT attempt_staging_cleanup_kind_check,
+    DROP CONSTRAINT attempt_staging_cleanup_identity_shape_check,
+    ADD CONSTRAINT attempt_staging_cleanup_kind_check
+        CHECK (kind IN ('write_intent', 'object', 'multipart')),
+    ADD CONSTRAINT attempt_staging_cleanup_identity_shape_check CHECK (
+        (kind IN ('write_intent', 'object') AND upload_id IS NULL)
+        OR (kind = 'multipart' AND version_id IS NULL AND upload_id IS NOT NULL)
+    );

crates/remora-attempt/src/lib.rs

@@ -10,13 +10,13 @@ mod types;
 pub use error::AttemptError;
 pub use migrate::migrate;
 pub use mutation_types::{
-    AttemptMutationOutcome, CheckpointProjection, CheckpointRecord, CommittedManifestRecord,
+    AttemptMutationOutcome, CheckpointMultipart, CheckpointProjection, CheckpointRecord,
-    CommittedResultRecord, DurableDescriptor, FailureRecord, JobMutationProjection,
+    CommittedManifestRecord, CommittedResultRecord, DurableDescriptor, FailureRecord,
-    MutationRejection, ProgressRecord,
+    JobMutationProjection, MutationRejection, ProgressRecord,
 };
 pub use recovery_types::{
     CancelOutcome, CleanupClaim, CleanupCompletion, CleanupIdentity, CooperativeStopOutcome,
-    ReconcileReport, RecoveryConfig, StagingMultipart, StagingObject,
+    ReconcileReport, RecoveryConfig, StagingMultipart, StagingObject, StagingWriteIntent,
 };
 pub use service::{AttemptConfig, AttemptService};
 pub use types::{

crates/remora-attempt/src/mutation_types.rs

@@ -18,6 +18,16 @@ pub struct CheckpointRecord {
     pub resolved_source_id: OpaqueId,
     pub validated_bytes: u64,
     pub validated_members: u64,
+    #[serde(default)]
+    pub staging_multipart: Option<CheckpointMultipart>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+pub struct CheckpointMultipart {
+    pub target_store: String,
+    pub bucket: String,
+    pub key: String,
+    pub upload_id: String,
 }
 
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]

crates/remora-attempt/src/mutations.rs

@@ -73,7 +73,11 @@ impl AttemptService {
         let outcome = self
             .publish_success(authority, operation_id, result)
             .await?;
-        self.publication_outcome(operation_id, outcome).await
+        let publication = self.publication_outcome(operation_id, outcome).await?;
+        if matches!(publication, OutcomePublication::Accepted { .. }) {
+            self.release_exact(authority).await;
+        }
+        Ok(publication)
     }
 
     pub async fn persist_checkpoint(
@@ -118,13 +122,16 @@ impl AttemptService {
                     }
                     let now = OffsetDateTime::now_utc();
                     sqlx::query(
-                        "UPDATE jobs SET checkpoint = $2, revision = revision + 1,
+                        "UPDATE jobs
-                            updated_at = $3 WHERE job_id = $1",
+                         SET checkpoint = $2, checkpoint_owner_attempt_id = $3,
+                             revision = revision + 1, updated_at = $4
+                         WHERE job_id = $1",
                     )
                     .bind(authority.job_id.as_str())
                     .bind(Json(&CheckpointProjection::Reusable {
                         checkpoint: checkpoint.clone(),
                     }))
+                    .bind(authority.attempt_id.as_str())
                     .bind(now)
                     .execute(tx.as_mut())
                     .await?;
@@ -143,7 +150,11 @@ impl AttemptService {
         class: RuntimeFailureClass,
     ) -> Result<OutcomePublication, AttemptError> {
         let outcome = self.publish_failure(authority, operation_id, class).await?;
-        self.publication_outcome(operation_id, outcome).await
+        let publication = self.publication_outcome(operation_id, outcome).await?;
+        if matches!(publication, OutcomePublication::Accepted { .. }) {
+            self.release_exact(authority).await;
+        }
+        Ok(publication)
     }
 
     pub async fn report_progress(
@@ -227,12 +238,14 @@ impl AttemptService {
                     };
                     let changed = sqlx::query(
                         "UPDATE jobs SET phase = 'succeeded', authority_attempt_id = NULL,
-                            checkpoint = $2, committed_result = $3, terminal_at = $4,
+                            checkpoint = $2, checkpoint_owner_attempt_id = $3,
-                            revision = revision + 1, updated_at = $4
+                            committed_result = $4, terminal_at = $5,
+                            revision = revision + 1, updated_at = $5
                          WHERE job_id = $1 AND phase = 'running' AND cancel_requested = FALSE",
                     )
                     .bind(authority.job_id.as_str())
                     .bind(Json(&complete_checkpoint))
+                    .bind(authority.attempt_id.as_str())
                     .bind(stored)
                     .bind(now)
                     .execute(tx.as_mut())
@@ -544,11 +557,26 @@ fn validate_checkpoint(checkpoint: &CheckpointRecord) -> Result<(), AttemptError
     if (checkpoint.validated_bytes == 0 && checkpoint.validated_members == 0)
         || checkpoint.validated_bytes > MAX_SAFE_INTEGER
         || checkpoint.validated_members > MAX_SAFE_INTEGER
+        || (checkpoint.validated_members == 0) != checkpoint.staging_multipart.is_some()
     {
         return Err(AttemptError::Invalid(
             "checkpoint exceeds the safe-integer range".into(),
         ));
     }
+    if checkpoint
+        .staging_multipart
+        .as_ref()
+        .is_some_and(|identity| {
+            identity.target_store.is_empty()
+                || identity.bucket.is_empty()
+                || identity.key.is_empty()
+                || identity.upload_id.is_empty()
+        })
+    {
+        return Err(AttemptError::Invalid(
+            "checkpoint multipart identity is invalid".into(),
+        ));
+    }
     Ok(())
 }
 

crates/remora-attempt/src/recovery.rs

@@ -1,9 +1,14 @@
 use std::time::Duration;
 
 use remora_coordination::{LeaseOwner, OpCode};
-use remora_domain::{FailureClass, FailureDisposition, JobPhase, OpaqueId};
+use remora_domain::{
-use remora_durable::{DurableError, enqueue, with_transaction};
+    FailureClass, FailureDisposition, JobPhase, OpaqueId, fingerprint, ports::MutationScope,
-use remora_target::validate_attempt_key;
+};
+use remora_durable::{
+    DurableError, MutationDisposition, MutationStatus, enqueue, execute_mutation, with_transaction,
+};
+use remora_target::{attempt_key_attempt_id, attempt_staging_prefix, validate_attempt_key};
+use serde::Serialize;
 use serde_json::json;
 use sqlx::{FromRow, Postgres, Transaction, types::Json};
 use time::OffsetDateTime;
@@ -12,7 +17,7 @@ use crate::service::new_id;
 use crate::{
     AttemptAuthority, AttemptError, AttemptService, CancelOutcome, CleanupClaim, CleanupCompletion,
     CleanupIdentity, CooperativeStopOutcome, FailureRecord, ReconcileReport, RecoveryConfig,
-    StagingMultipart, StagingObject,
+    StagingMultipart, StagingObject, StagingWriteIntent,
 };
 
 impl AttemptService {
@@ -73,57 +78,79 @@ impl AttemptService {
         &self,
         authority: &AttemptAuthority,
     ) -> Result<CooperativeStopOutcome, AttemptError> {
-        let row = sqlx::query_as::<_, StopRow>(
+        let authority_for_effect = authority.clone();
-            "SELECT a.job_id, a.fence, a.worker_incarnation, a.state,
+        let coordination = self.coordination.clone();
-                    j.authority_attempt_id, j.cancel_requested
+        let outcome = with_transaction(&self.pool, move |tx| {
-             FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+            Box::pin(async move {
-             WHERE a.attempt_id = $1",
+                cooperative_stop_transaction(tx, &coordination, &authority_for_effect).await
-        )
+            })
-        .bind(authority.attempt_id.as_str())
+        })
-        .fetch_optional(&self.pool)
         .await?;
-        let Some(row) = row else {
+        if outcome == CooperativeStopOutcome::Stopped {
-            return Ok(CooperativeStopOutcome::AuthorityMismatch);
+            let _ = self
-        };
+                .coordination
-        if row.job_id != authority.job_id.as_str()
+                .release(authority.job_id.as_str(), &lease_owner(authority))
-            || u64::try_from(row.fence).ok() != Some(authority.fence)
+                .await;
-            || row.worker_incarnation != authority.worker_incarnation.as_str()
+            self.release_local(&authority.job_id);
-        {
-            return Ok(CooperativeStopOutcome::AuthorityMismatch);
         }
-        if row.state == "stopped" {
+        Ok(outcome)
-            return Ok(CooperativeStopOutcome::Stopped);
+    }
-        }
+
-        if row.state != "executing" {
+    pub async fn cooperative_stop_operation(
-            return Ok(CooperativeStopOutcome::StopNotRequired);
+        &self,
-        }
+        authority: &AttemptAuthority,
-        let current = row.authority_attempt_id.as_deref() == Some(authority.attempt_id.as_str());
+        operation_id: &OpaqueId,
-        let live = self
+    ) -> Result<CooperativeStopOutcome, AttemptError> {
-            .coordination
+        let request_fingerprint = fingerprint(&StopRequest {
-            .get_live_lease(authority.job_id.as_str())
+            job_id: &authority.job_id,
-            .await?
+            attempt_id: &authority.attempt_id,
-            .is_some_and(|lease| lease.record.owner() == lease_owner(authority));
+            fence: authority.fence,
-        if current && live && !row.cancel_requested {
+            worker_incarnation: &authority.worker_incarnation,
-            return Ok(CooperativeStopOutcome::StopNotRequired);
+        })
-        }
+        .map_err(|error| AttemptError::Invalid(error.to_string()))?;
-        let now = OffsetDateTime::now_utc();
+        let authority_for_effect = authority.clone();
-        let changed = sqlx::query(
+        let coordination = self.coordination.clone();
-            "UPDATE attempts SET state = 'stopped', stopped_at = $2
+        let outcome = execute_mutation(
-             WHERE attempt_id = $1 AND state IN ('authorized', 'executing')",
+            &self.pool,
+            MutationScope {
+                subject: &self.operation_subject,
+                operation: "attempt.mutation",
+                key: operation_id.as_str(),
+            },
+            "cooperative_stop",
+            &[],
+            request_fingerprint,
+            move |tx| {
+                Box::pin(async move {
+                    let decision =
+                        cooperative_stop_transaction(tx, &coordination, &authority_for_effect)
+                            .await?;
+                    let status = if decision == CooperativeStopOutcome::AuthorityMismatch {
+                        MutationStatus::Rejected
+                    } else {
+                        MutationStatus::Accepted
+                    };
+                    Ok((
+                        status,
+                        serde_json::to_value(decision).map_err(DurableError::effect)?,
+                    ))
+                })
+            },
         )
-        .bind(authority.attempt_id.as_str())
-        .bind(now)
-        .execute(&self.pool)
         .await?;
-        if changed.rows_affected() == 0 {
+        if outcome.disposition == MutationDisposition::Conflict {
-            return Ok(CooperativeStopOutcome::StopNotRequired);
+            return Err(AttemptError::OperationConflict);
         }
-        let _ = self
+        let decision = serde_json::from_value(outcome.response)
-            .coordination
+            .map_err(|error| AttemptError::Invalid(error.to_string()))?;
-            .release(authority.job_id.as_str(), &lease_owner(authority))
+        if decision == CooperativeStopOutcome::Stopped {
-            .await;
+            let _ = self
-        self.release_local(&authority.job_id);
+                .coordination
-        Ok(CooperativeStopOutcome::Stopped)
+                .release(authority.job_id.as_str(), &lease_owner(authority))
+                .await;
+            self.release_local(&authority.job_id);
+        }
+        Ok(decision)
     }
 
     pub async fn mark_worker_down(
@@ -146,7 +173,7 @@ impl AttemptService {
         attempt_id: &OpaqueId,
         object: &StagingObject,
     ) -> Result<(), AttemptError> {
-        validate_staging_key(attempt_id, &object.key)?;
+        validate_staging_key(&self.pool, attempt_id, &object.key).await?;
         insert_cleanup_identity(
             &self.pool,
             attempt_id,
@@ -162,12 +189,105 @@ impl AttemptService {
         .await
     }
 
+    pub async fn register_staging_write_intent_operation(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        intent: StagingWriteIntent,
+    ) -> Result<bool, AttemptError> {
+        self.register_staging_operation(
+            authority,
+            operation_id,
+            "register_staging_write_intent",
+            intent.clone(),
+            move |tx, attempt_id| {
+                Box::pin(async move {
+                    validate_staging_key_transaction(tx, &attempt_id, &intent.key).await?;
+                    insert_cleanup_identity_transaction(
+                        tx,
+                        &attempt_id,
+                        NewCleanupIdentity {
+                            kind: "write_intent",
+                            target_store: &intent.target_store,
+                            bucket: &intent.bucket,
+                            key: &intent.key,
+                            version_id: None,
+                            upload_id: None,
+                        },
+                    )
+                    .await
+                })
+            },
+        )
+        .await
+    }
+
+    pub async fn register_staging_object_operation(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        object: StagingObject,
+        completed_upload_id: Option<String>,
+    ) -> Result<bool, AttemptError> {
+        self.register_staging_operation(
+            authority,
+            operation_id,
+            "register_staging_object",
+            StagingObjectRegistration {
+                object: object.clone(),
+                completed_upload_id: completed_upload_id.clone(),
+            },
+            move |tx, attempt_id| {
+                Box::pin(async move {
+                    validate_staging_key_transaction(tx, &attempt_id, &object.key).await?;
+                    insert_cleanup_identity_transaction(
+                        tx,
+                        &attempt_id,
+                        NewCleanupIdentity {
+                            kind: "object",
+                            target_store: &object.target_store,
+                            bucket: &object.bucket,
+                            key: &object.key,
+                            version_id: object.version_id.as_deref(),
+                            upload_id: None,
+                        },
+                    )
+                    .await?;
+                    complete_pending_identity(
+                        tx,
+                        &attempt_id,
+                        "write_intent",
+                        &object.target_store,
+                        &object.bucket,
+                        &object.key,
+                        None,
+                    )
+                    .await?;
+                    if let Some(upload_id) = completed_upload_id {
+                        complete_pending_identity(
+                            tx,
+                            &attempt_id,
+                            "multipart",
+                            &object.target_store,
+                            &object.bucket,
+                            &object.key,
+                            Some(&upload_id),
+                        )
+                        .await?;
+                    }
+                    Ok(())
+                })
+            },
+        )
+        .await
+    }
+
     pub async fn register_staging_multipart(
         &self,
         attempt_id: &OpaqueId,
         multipart: &StagingMultipart,
     ) -> Result<(), AttemptError> {
-        validate_staging_key(attempt_id, &multipart.key)?;
+        validate_staging_key(&self.pool, attempt_id, &multipart.key).await?;
         if multipart.upload_id.is_empty() {
             return Err(AttemptError::Invalid(
                 "multipart upload identity cannot be empty".into(),
@@ -188,6 +308,107 @@ impl AttemptService {
         .await
     }
 
+    pub async fn register_staging_multipart_operation(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        multipart: StagingMultipart,
+    ) -> Result<bool, AttemptError> {
+        if multipart.upload_id.is_empty() {
+            return Err(AttemptError::Invalid(
+                "multipart upload identity cannot be empty".into(),
+            ));
+        }
+        self.register_staging_operation(
+            authority,
+            operation_id,
+            "register_staging_multipart",
+            multipart.clone(),
+            move |tx, attempt_id| {
+                Box::pin(async move {
+                    validate_staging_key_transaction(tx, &attempt_id, &multipart.key).await?;
+                    insert_cleanup_identity_transaction(
+                        tx,
+                        &attempt_id,
+                        NewCleanupIdentity {
+                            kind: "multipart",
+                            target_store: &multipart.target_store,
+                            bucket: &multipart.bucket,
+                            key: &multipart.key,
+                            version_id: None,
+                            upload_id: Some(&multipart.upload_id),
+                        },
+                    )
+                    .await
+                })
+            },
+        )
+        .await
+    }
+
+    async fn register_staging_operation<T, F>(
+        &self,
+        authority: &AttemptAuthority,
+        operation_id: &OpaqueId,
+        kind: &'static str,
+        request: T,
+        effect: F,
+    ) -> Result<bool, AttemptError>
+    where
+        T: Serialize,
+        F: for<'c> FnOnce(
+                &'c mut remora_durable::PgTx<'_>,
+                OpaqueId,
+            ) -> remora_durable::BoxFuture<'c, Result<(), DurableError>>
+            + Send
+            + 'static,
+    {
+        let request_fingerprint = fingerprint(&RegistrationRequest {
+            job_id: &authority.job_id,
+            attempt_id: &authority.attempt_id,
+            fence: authority.fence,
+            worker_incarnation: &authority.worker_incarnation,
+            value: request,
+        })
+        .map_err(|error| AttemptError::Invalid(error.to_string()))?;
+        let authority = authority.clone();
+        let outcome = execute_mutation(
+            &self.pool,
+            MutationScope {
+                subject: &self.operation_subject,
+                operation: "attempt.mutation",
+                key: operation_id.as_str(),
+            },
+            kind,
+            &[],
+            request_fingerprint,
+            move |tx| {
+                Box::pin(async move {
+                    let eligible = lock_registration_authority(tx, &authority).await?;
+                    if eligible {
+                        effect(tx, authority.attempt_id).await?;
+                    }
+                    Ok((
+                        if eligible {
+                            MutationStatus::Accepted
+                        } else {
+                            MutationStatus::Rejected
+                        },
+                        serde_json::Value::Bool(eligible),
+                    ))
+                })
+            },
+        )
+        .await?;
+        if outcome.disposition == MutationDisposition::Conflict {
+            return Err(AttemptError::OperationConflict);
+        }
+        outcome
+            .response
+            .as_bool()
+            .ok_or_else(|| AttemptError::Invalid("invalid staging operation result".into()))
+    }
+
     pub async fn reconcile_once(
         &self,
         config: &RecoveryConfig,
@@ -235,6 +456,8 @@ impl AttemptService {
                 WHERE (c.state = 'pending'
                        OR (c.state = 'claimed' AND c.claimed_until <= $1))
                   AND j.authority_attempt_id IS DISTINCT FROM a.attempt_id
+                  AND (j.phase IN ('succeeded', 'failed', 'cancelled')
+                       OR j.checkpoint_owner_attempt_id IS DISTINCT FROM a.attempt_id)
                   AND COALESCE(j.committed_result ->> 'attempt_id', '') <> a.attempt_id
                   AND (a.state IN ('stopped', 'reaped') OR a.worker_down_at IS NOT NULL)
                 ORDER BY c.cleanup_id
@@ -279,6 +502,104 @@ impl AttemptService {
         })
     }
 
+    pub async fn reconcile_write_intent(
+        &self,
+        claim: &CleanupClaim,
+        objects: Vec<StagingObject>,
+        multiparts: Vec<StagingMultipart>,
+    ) -> Result<CleanupCompletion, AttemptError> {
+        let CleanupIdentity::WriteIntent(intent) = &claim.identity else {
+            return Err(AttemptError::Invalid(
+                "cleanup claim is not a staging write intent".into(),
+            ));
+        };
+        for object in &objects {
+            if object.target_store != intent.target_store
+                || object.bucket != intent.bucket
+                || object.key != intent.key
+            {
+                return Err(AttemptError::Invalid(
+                    "discovered object does not match the write intent".into(),
+                ));
+            }
+        }
+        for multipart in &multiparts {
+            if multipart.target_store != intent.target_store
+                || multipart.bucket != intent.bucket
+                || multipart.key != intent.key
+                || multipart.upload_id.is_empty()
+            {
+                return Err(AttemptError::Invalid(
+                    "discovered multipart does not match the write intent".into(),
+                ));
+            }
+        }
+        let cleanup_id = i64::try_from(claim.cleanup_id).map_err(invalid_number)?;
+        let claim_token = claim.claim_token.clone();
+        let attempt_id = claim.attempt_id.clone();
+        with_transaction(&self.pool, move |tx| {
+            Box::pin(async move {
+                let retained = sqlx::query_scalar::<_, i64>(
+                    "SELECT cleanup_id FROM attempt_staging_cleanup
+                     WHERE cleanup_id = $1 AND attempt_id = $2 AND kind = 'write_intent'
+                       AND state = 'claimed' AND claim_token = $3
+                     FOR UPDATE",
+                )
+                .bind(cleanup_id)
+                .bind(attempt_id.as_str())
+                .bind(claim_token.as_str())
+                .fetch_optional(tx.as_mut())
+                .await?;
+                if retained.is_none() {
+                    return Ok(CleanupCompletion::ClaimLost);
+                }
+                for object in &objects {
+                    insert_cleanup_identity_transaction(
+                        tx,
+                        &attempt_id,
+                        NewCleanupIdentity {
+                            kind: "object",
+                            target_store: &object.target_store,
+                            bucket: &object.bucket,
+                            key: &object.key,
+                            version_id: object.version_id.as_deref(),
+                            upload_id: None,
+                        },
+                    )
+                    .await?;
+                }
+                for multipart in &multiparts {
+                    insert_cleanup_identity_transaction(
+                        tx,
+                        &attempt_id,
+                        NewCleanupIdentity {
+                            kind: "multipart",
+                            target_store: &multipart.target_store,
+                            bucket: &multipart.bucket,
+                            key: &multipart.key,
+                            version_id: None,
+                            upload_id: Some(&multipart.upload_id),
+                        },
+                    )
+                    .await?;
+                }
+                sqlx::query(
+                    "UPDATE attempt_staging_cleanup
+                     SET state = 'complete', claim_token = NULL, claimed_until = NULL,
+                         completed_at = $2, last_error = NULL
+                     WHERE cleanup_id = $1",
+                )
+                .bind(cleanup_id)
+                .bind(OffsetDateTime::now_utc())
+                .execute(tx.as_mut())
+                .await?;
+                Ok(CleanupCompletion::Completed)
+            })
+        })
+        .await
+        .map_err(Into::into)
+    }
+
     pub async fn retry_cleanup(
         &self,
         claim: &CleanupClaim,
@@ -499,6 +820,8 @@ impl AttemptService {
              JOIN attempts a ON a.attempt_id = m.attempt_id
              JOIN jobs j ON j.job_id = a.job_id
              WHERE j.authority_attempt_id IS DISTINCT FROM a.attempt_id
+               AND (j.phase IN ('succeeded', 'failed', 'cancelled')
+                    OR j.checkpoint_owner_attempt_id IS DISTINCT FROM a.attempt_id)
                AND COALESCE(j.committed_result ->> 'attempt_id', '') <> a.attempt_id
                AND (a.state IN ('stopped', 'reaped') OR a.worker_down_at IS NOT NULL)
              ON CONFLICT DO NOTHING",
@@ -517,6 +840,8 @@ impl AttemptService {
              JOIN jobs j ON j.job_id = a.job_id
              WHERE r.state = 'sealed'
                AND j.authority_attempt_id IS DISTINCT FROM a.attempt_id
+               AND (j.phase IN ('succeeded', 'failed', 'cancelled')
+                    OR j.checkpoint_owner_attempt_id IS DISTINCT FROM a.attempt_id)
                AND COALESCE(j.committed_result ->> 'attempt_id', '') <> a.attempt_id
                AND (a.state IN ('stopped', 'reaped') OR a.worker_down_at IS NOT NULL)
              ON CONFLICT DO NOTHING",
@@ -544,6 +869,199 @@ impl AttemptService {
     }
 }
 
+async fn cooperative_stop_transaction(
+    tx: &mut remora_durable::PgTx<'_>,
+    coordination: &remora_coordination::CoordinationStore,
+    authority: &AttemptAuthority,
+) -> Result<CooperativeStopOutcome, DurableError> {
+    let row = sqlx::query_as::<_, StopRow>(
+        "SELECT a.job_id, a.fence, a.worker_incarnation, a.state,
+                j.authority_attempt_id, j.cancel_requested
+         FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.attempt_id = $1
+         FOR UPDATE OF a, j",
+    )
+    .bind(authority.attempt_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?;
+    let Some(row) = row else {
+        return Ok(CooperativeStopOutcome::AuthorityMismatch);
+    };
+    if row.job_id != authority.job_id.as_str()
+        || u64::try_from(row.fence).ok() != Some(authority.fence)
+        || row.worker_incarnation != authority.worker_incarnation.as_str()
+    {
+        return Ok(CooperativeStopOutcome::AuthorityMismatch);
+    }
+    if row.state == "stopped" {
+        return Ok(CooperativeStopOutcome::Stopped);
+    }
+    if row.state != "executing" {
+        return Ok(CooperativeStopOutcome::StopNotRequired);
+    }
+    let current = row.authority_attempt_id.as_deref() == Some(authority.attempt_id.as_str());
+    let live = coordination
+        .get_live_lease(authority.job_id.as_str())
+        .await
+        .map_err(DurableError::effect)?
+        .is_some_and(|lease| lease.record.owner() == lease_owner(authority));
+    if current && live && !row.cancel_requested {
+        return Ok(CooperativeStopOutcome::StopNotRequired);
+    }
+    let changed = sqlx::query(
+        "UPDATE attempts SET state = 'stopped', stopped_at = $2
+         WHERE attempt_id = $1 AND state IN ('authorized', 'executing')",
+    )
+    .bind(authority.attempt_id.as_str())
+    .bind(OffsetDateTime::now_utc())
+    .execute(tx.as_mut())
+    .await?;
+    Ok(if changed.rows_affected() == 1 {
+        CooperativeStopOutcome::Stopped
+    } else {
+        CooperativeStopOutcome::StopNotRequired
+    })
+}
+
+async fn lock_registration_authority(
+    tx: &mut remora_durable::PgTx<'_>,
+    authority: &AttemptAuthority,
+) -> Result<bool, DurableError> {
+    let row = sqlx::query_as::<_, RegistrationAuthorityRow>(
+        "SELECT a.job_id, a.fence, a.worker_incarnation, a.state,
+                j.authority_attempt_id, j.cancel_requested
+         FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.attempt_id = $1
+         FOR UPDATE OF a, j",
+    )
+    .bind(authority.attempt_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?;
+    Ok(row.is_some_and(|row| {
+        row.job_id == authority.job_id.as_str()
+            && u64::try_from(row.fence).ok() == Some(authority.fence)
+            && row.worker_incarnation == authority.worker_incarnation.as_str()
+            && row.state == "executing"
+            && row.authority_attempt_id.as_deref() == Some(authority.attempt_id.as_str())
+            && !row.cancel_requested
+    }))
+}
+
+async fn validate_staging_key_transaction(
+    tx: &mut remora_durable::PgTx<'_>,
+    attempt_id: &OpaqueId,
+    key: &str,
+) -> Result<(), DurableError> {
+    validate_attempt_key(key).map_err(DurableError::effect)?;
+    let (staging_origin, target_prefix) = sqlx::query_as::<_, (String, Json<Vec<String>>)>(
+        "SELECT a.staging_origin_attempt_id, j.target_prefix
+             FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+             WHERE a.attempt_id = $1",
+    )
+    .bind(attempt_id.as_str())
+    .fetch_one(tx.as_mut())
+    .await?;
+    let staging_origin = OpaqueId::parse(staging_origin).map_err(DurableError::effect)?;
+    if attempt_key_attempt_id(key).map_err(DurableError::effect)? != staging_origin
+        || !key.starts_with(&attempt_staging_prefix(
+            &target_prefix.0.join("/"),
+            &staging_origin,
+        ))
+    {
+        return Err(DurableError::effect(
+            "staging key belongs to a different attempt",
+        ));
+    }
+    Ok(())
+}
+
+async fn insert_cleanup_identity_transaction(
+    tx: &mut remora_durable::PgTx<'_>,
+    attempt_id: &OpaqueId,
+    identity: NewCleanupIdentity<'_>,
+) -> Result<(), DurableError> {
+    if identity.target_store.is_empty() || identity.bucket.is_empty() {
+        return Err(DurableError::effect(
+            "staging target store and bucket cannot be empty",
+        ));
+    }
+    let changed = sqlx::query(
+        "INSERT INTO attempt_staging_cleanup (
+            attempt_id, kind, target_store, bucket, object_key, version_id, upload_id, created_at
+         )
+         SELECT $1, $2, $3, $4, $5, $6, $7, $8
+         WHERE EXISTS (
+             SELECT 1 FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+             WHERE a.attempt_id = $1 AND j.target_store = $3 AND j.target_bucket = $4
+         )
+         ON CONFLICT DO NOTHING",
+    )
+    .bind(attempt_id.as_str())
+    .bind(identity.kind)
+    .bind(identity.target_store)
+    .bind(identity.bucket)
+    .bind(identity.key)
+    .bind(identity.version_id)
+    .bind(identity.upload_id)
+    .bind(OffsetDateTime::now_utc())
+    .execute(tx.as_mut())
+    .await?;
+    if changed.rows_affected() == 0 {
+        let binding = sqlx::query_as::<_, (String, String)>(
+            "SELECT j.target_store, j.target_bucket
+             FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+             WHERE a.attempt_id = $1",
+        )
+        .bind(attempt_id.as_str())
+        .fetch_optional(tx.as_mut())
+        .await?;
+        match binding {
+            None => return Err(DurableError::effect("attempt does not exist")),
+            Some((store, bucket))
+                if store != identity.target_store || bucket != identity.bucket =>
+            {
+                return Err(DurableError::effect(
+                    "staging identity does not match the attempt target",
+                ));
+            }
+            Some(_) => {}
+        }
+    }
+    Ok(())
+}
+
+async fn complete_pending_identity(
+    tx: &mut remora_durable::PgTx<'_>,
+    attempt_id: &OpaqueId,
+    kind: &str,
+    target_store: &str,
+    bucket: &str,
+    key: &str,
+    upload_id: Option<&str>,
+) -> Result<(), DurableError> {
+    let changed = sqlx::query(
+        "UPDATE attempt_staging_cleanup
+         SET state = 'complete', completed_at = $7
+         WHERE attempt_id = $1 AND kind = $2 AND target_store = $3 AND bucket = $4
+           AND object_key = $5 AND upload_id IS NOT DISTINCT FROM $6 AND state = 'pending'",
+    )
+    .bind(attempt_id.as_str())
+    .bind(kind)
+    .bind(target_store)
+    .bind(bucket)
+    .bind(key)
+    .bind(upload_id)
+    .bind(OffsetDateTime::now_utc())
+    .execute(tx.as_mut())
+    .await?;
+    if changed.rows_affected() != 1 {
+        return Err(DurableError::effect(
+            "staging write lifecycle identity is missing or unavailable",
+        ));
+    }
+    Ok(())
+}
+
 async fn insert_cleanup_identity(
     pool: &sqlx::PgPool,
     attempt_id: &OpaqueId,
@@ -662,6 +1180,7 @@ async fn reap_lost_attempt(
     let inactive = match row.state.as_str() {
         "authorized" => reservation.is_none() && !exact_live,
         "executing" => row.worker_down_at.is_some() || !exact_live,
+        "stopped" => !exact_live,
         _ => false,
     };
     if !inactive {
@@ -718,9 +1237,30 @@ async fn reap_lost_attempt(
     })
 }
 
-fn validate_staging_key(attempt_id: &OpaqueId, key: &str) -> Result<(), AttemptError> {
+async fn validate_staging_key(
+    pool: &sqlx::PgPool,
+    attempt_id: &OpaqueId,
+    key: &str,
+) -> Result<(), AttemptError> {
     validate_attempt_key(key).map_err(|error| AttemptError::Invalid(error.to_string()))?;
-    if key.split('/').nth(2) != Some(attempt_id.as_str()) {
+    let (staging_origin, target_prefix) = sqlx::query_as::<_, (String, Json<Vec<String>>)>(
+        "SELECT a.staging_origin_attempt_id, j.target_prefix
+         FROM attempts a JOIN jobs j ON j.job_id = a.job_id
+         WHERE a.attempt_id = $1",
+    )
+    .bind(attempt_id.as_str())
+    .fetch_optional(pool)
+    .await?
+    .ok_or_else(|| AttemptError::Invalid("attempt does not exist".into()))?;
+    let staging_origin = OpaqueId::parse(staging_origin)
+        .map_err(|error| AttemptError::Invalid(error.to_string()))?;
+    if attempt_key_attempt_id(key).map_err(|error| AttemptError::Invalid(error.to_string()))?
+        != staging_origin
+        || !key.starts_with(&attempt_staging_prefix(
+            &target_prefix.0.join("/"),
+            &staging_origin,
+        ))
+    {
         return Err(AttemptError::Invalid(
             "staging key belongs to a different attempt".into(),
         ));
@@ -788,6 +1328,39 @@ struct StopRow {
     cancel_requested: bool,
 }
 
+#[derive(FromRow)]
+struct RegistrationAuthorityRow {
+    job_id: String,
+    fence: i64,
+    worker_incarnation: String,
+    state: String,
+    authority_attempt_id: Option<String>,
+    cancel_requested: bool,
+}
+
+#[derive(Serialize)]
+struct StopRequest<'a> {
+    job_id: &'a OpaqueId,
+    attempt_id: &'a OpaqueId,
+    fence: u64,
+    worker_incarnation: &'a OpaqueId,
+}
+
+#[derive(Serialize)]
+struct RegistrationRequest<'a, T> {
+    job_id: &'a OpaqueId,
+    attempt_id: &'a OpaqueId,
+    fence: u64,
+    worker_incarnation: &'a OpaqueId,
+    value: T,
+}
+
+#[derive(Serialize)]
+struct StagingObjectRegistration {
+    object: StagingObject,
+    completed_upload_id: Option<String>,
+}
+
 #[derive(FromRow)]
 struct RecoveryJobRow {
     phase: String,
@@ -827,6 +1400,11 @@ struct CleanupRow {
 impl CleanupRow {
     fn try_into_claim(self, claim_token: OpaqueId) -> Result<CleanupClaim, AttemptError> {
         let identity = match self.kind.as_str() {
+            "write_intent" => CleanupIdentity::WriteIntent(StagingWriteIntent {
+                target_store: self.target_store,
+                bucket: self.bucket,
+                key: self.object_key,
+            }),
             "object" => CleanupIdentity::Object(StagingObject {
                 target_store: self.target_store,
                 bucket: self.bucket,

crates/remora-attempt/src/recovery_types.rs

@@ -1,6 +1,7 @@
 use std::time::Duration;
 
 use remora_domain::OpaqueId;
+use serde::{Deserialize, Serialize};
 
 #[derive(Clone, Debug)]
 pub struct RecoveryConfig {
@@ -27,7 +28,8 @@ pub enum CancelOutcome {
     NotFound,
 }
 
-#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
 pub enum CooperativeStopOutcome {
     Stopped,
     StopNotRequired,
@@ -59,7 +61,7 @@ impl ReconcileReport {
     }
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
 pub struct StagingObject {
     pub target_store: String,
     pub bucket: String,
@@ -67,7 +69,7 @@ pub struct StagingObject {
     pub version_id: Option<String>,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq)]
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
 pub struct StagingMultipart {
     pub target_store: String,
     pub bucket: String,
@@ -75,8 +77,16 @@ pub struct StagingMultipart {
     pub upload_id: String,
 }
 
+#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
+pub struct StagingWriteIntent {
+    pub target_store: String,
+    pub bucket: String,
+    pub key: String,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub enum CleanupIdentity {
+    WriteIntent(StagingWriteIntent),
     Object(StagingObject),
     Multipart(StagingMultipart),
 }

crates/remora-attempt/src/service.rs

@@ -559,7 +559,7 @@ impl AttemptService {
         self.local_permits.release(job_id.as_str());
     }
 
-    async fn release_exact(&self, authority: &AttemptAuthority) {
+    pub(crate) async fn release_exact(&self, authority: &AttemptAuthority) {
         if let Ok(owner) = lease_owner(authority) {
             let _ = self
                 .coordination
@@ -662,16 +662,25 @@ async fn authorize_attempt(
             Option<String>,
             Option<Json<crate::DurableDescriptor>>,
             Option<Json<CheckpointProjection>>,
+            Option<String>,
         ),
     >(
         "SELECT phase, dispatch_needed, current_fence, authority_attempt_id,
-                resolved_source, checkpoint
+                resolved_source, checkpoint, checkpoint_owner_attempt_id
          FROM jobs WHERE job_id = $1 FOR UPDATE",
     )
     .bind(input.job_id.as_str())
     .fetch_optional(tx.as_mut())
     .await?;
-    let Some((phase, dispatch_needed, current_fence, authority, resolved_source, checkpoint)) = row
+    let Some((
+        phase,
+        dispatch_needed,
+        current_fence,
+        authority,
+        resolved_source,
+        checkpoint,
+        checkpoint_owner,
+    )) = row
     else {
         return Ok(claim_rejected(ClaimRejection::JobUnavailable));
     };
@@ -682,6 +691,26 @@ async fn authorize_attempt(
     if current_fence >= input.max_attempts {
         return Ok(claim_rejected(ClaimRejection::AttemptsExhausted));
     }
+    let reusable = resolved_source
+        .as_ref()
+        .zip(checkpoint.as_ref())
+        .and_then(|(source, checkpoint)| checkpoint.0.reusable().map(|value| (source, value)))
+        .filter(|(source, checkpoint)| checkpoint.resolved_source_id == source.0.id);
+    if let (Some(_), Some(owner)) = (reusable, checkpoint_owner.as_deref()) {
+        let stopped = sqlx::query_scalar::<_, bool>(
+            "SELECT state = 'stopped'
+             FROM attempts
+             WHERE attempt_id = $1 AND job_id = $2",
+        )
+        .bind(owner)
+        .bind(input.job_id.as_str())
+        .fetch_optional(tx.as_mut())
+        .await?
+        .unwrap_or(false);
+        if !stopped {
+            return Ok(claim_rejected(ClaimRejection::JobUnavailable));
+        }
+    }
     input.faults.hit(AttemptFaultPoint::ClaimBeforeReserve);
     let reservation = input
         .coordination
@@ -697,26 +726,47 @@ async fn authorize_attempt(
     input.reserved.store(true, Ordering::Release);
     input.faults.hit(AttemptFaultPoint::ClaimAfterReserve);
     let fence = current_fence + 1;
-    let initial_progress = match (resolved_source, checkpoint) {
+    let checkpoint_stage = match (reusable, checkpoint_owner.as_deref()) {
-        (Some(source), Some(checkpoint))
+        (Some((_, checkpoint)), Some(owner)) if checkpoint.validated_bytes > 0 => {
-            if checkpoint
+            load_checkpoint_stage(tx, input.job_id, owner, checkpoint).await?
-                .0
-                .reusable()
-                .is_some_and(|value| value.resolved_source_id == source.0.id) =>
-        {
-            checkpoint.0.reusable().unwrap().validated_bytes
         }
-        _ => 0,
+        _ => None,
     };
+    if checkpoint.is_some() && checkpoint_stage.is_none() {
+        sqlx::query(
+            "UPDATE jobs
+             SET checkpoint = NULL, checkpoint_owner_attempt_id = NULL,
+                 revision = revision + 1, updated_at = $2
+             WHERE job_id = $1",
+        )
+        .bind(input.job_id.as_str())
+        .bind(OffsetDateTime::now_utc())
+        .execute(tx.as_mut())
+        .await?;
+    }
+    let initial_progress = checkpoint_stage
+        .as_ref()
+        .map_or(0, |stage| stage.validated_bytes);
+    let staging_origin = checkpoint_stage
+        .as_ref()
+        .map_or(input.attempt_id, |stage| &stage.staging_origin_attempt_id);
+    let checkpoint_owner_after_claim = checkpoint_stage.as_ref().map(|_| input.attempt_id.as_str());
     let now = OffsetDateTime::now_utc();
-    let envelope =
+    let envelope = load_execution_envelope_record(
-        load_execution_envelope_record(tx, input.job_id, input.attempt_id, fence, input.worker)
+        tx,
-            .await?;
+        input.job_id,
+        input.attempt_id,
+        fence,
+        input.worker,
+        staging_origin,
+    )
+    .await?;
     sqlx::query(
         "INSERT INTO attempts (
             attempt_id, job_id, fence, worker_incarnation, state,
-            initial_progress, authorized_at, executing_at, reaped_at, execution_envelope
+            initial_progress, authorized_at, executing_at, reaped_at, execution_envelope,
-         ) VALUES ($1, $2, $3, $4, 'authorized', $5, $6, NULL, NULL, $7)",
+            staging_origin_attempt_id
+         ) VALUES ($1, $2, $3, $4, 'authorized', $5, $6, NULL, NULL, $7, $8)",
     )
     .bind(input.attempt_id.as_str())
     .bind(input.job_id.as_str())
@@ -725,18 +775,24 @@ async fn authorize_attempt(
     .bind(i64::try_from(initial_progress).map_err(DurableError::effect)?)
     .bind(now)
     .bind(Json(&envelope))
+    .bind(staging_origin.as_str())
     .execute(tx.as_mut())
     .await?;
+    if let Some(stage) = checkpoint_stage {
+        transfer_checkpoint_stage(tx, &stage, input.attempt_id).await?;
+    }
     sqlx::query(
         "UPDATE jobs
          SET phase = 'running', current_fence = $2, authority_attempt_id = $3,
-             dispatch_needed = FALSE, revision = revision + 1, updated_at = $4
+             checkpoint_owner_attempt_id = $5, dispatch_needed = FALSE,
+             revision = revision + 1, updated_at = $4
          WHERE job_id = $1",
     )
     .bind(input.job_id.as_str())
     .bind(i64::try_from(fence).map_err(DurableError::effect)?)
     .bind(input.attempt_id.as_str())
     .bind(now)
+    .bind(checkpoint_owner_after_claim)
     .execute(tx.as_mut())
     .await?;
     Ok(ClaimOutcome::Claimed {
@@ -751,12 +807,277 @@ async fn authorize_attempt(
     })
 }
 
+struct CheckpointStage {
+    owner_attempt_id: OpaqueId,
+    staging_origin_attempt_id: OpaqueId,
+    validated_bytes: u64,
+    validated_members: u64,
+    inventory_bytes: u64,
+    staging_multipart: Option<crate::CheckpointMultipart>,
+}
+
+async fn load_checkpoint_stage(
+    tx: &mut remora_durable::PgTx<'_>,
+    job_id: &OpaqueId,
+    owner: &str,
+    checkpoint: &crate::CheckpointRecord,
+) -> Result<Option<CheckpointStage>, DurableError> {
+    let Some(staging_origin) = sqlx::query_scalar::<_, String>(
+        "SELECT staging_origin_attempt_id
+         FROM attempts
+         WHERE attempt_id = $1 AND job_id = $2 AND state = 'stopped'",
+    )
+    .bind(owner)
+    .bind(job_id.as_str())
+    .fetch_optional(tx.as_mut())
+    .await?
+    else {
+        return Ok(None);
+    };
+    let Some((state, member_count, inventory_origin, inventory_prefix, target_prefix)) =
+        sqlx::query_as::<
+            _,
+            (
+                String,
+                i64,
+                Option<String>,
+                Option<String>,
+                Json<Vec<String>>,
+            ),
+        >(
+            "SELECT i.state, i.member_count, i.staging_origin_attempt_id,
+                    i.target_prefix, j.target_prefix
+             FROM result_inventories i
+             JOIN attempts a ON a.attempt_id = i.attempt_id
+             JOIN jobs j ON j.job_id = a.job_id
+             WHERE i.attempt_id = $1
+             FOR UPDATE OF i",
+        )
+        .bind(owner)
+        .fetch_optional(tx.as_mut())
+        .await?
+    else {
+        return Ok(None);
+    };
+    if inventory_origin.as_deref() != Some(staging_origin.as_str())
+        || inventory_prefix.as_deref() != Some(target_prefix.0.join("/").as_str())
+    {
+        return Ok(None);
+    }
+    let validated_members =
+        i64::try_from(checkpoint.validated_members).map_err(DurableError::effect)?;
+    let validated_bytes =
+        i64::try_from(checkpoint.validated_bytes).map_err(DurableError::effect)?;
+    if state != "open" || member_count < validated_members {
+        return Ok(None);
+    }
+    let (retained_members, retained_bytes) = sqlx::query_as::<_, (i64, i64)>(
+        "SELECT COUNT(*), COALESCE(SUM(size), 0)::BIGINT
+         FROM (
+             SELECT size
+             FROM result_inventory_members
+             WHERE attempt_id = $1
+             ORDER BY path COLLATE \"C\", member_id COLLATE \"C\"
+             LIMIT $2
+         ) retained",
+    )
+    .bind(owner)
+    .bind(validated_members)
+    .fetch_one(tx.as_mut())
+    .await?;
+    if retained_members != validated_members || retained_bytes > validated_bytes {
+        return Ok(None);
+    }
+    let partial = retained_bytes < validated_bytes;
+    if partial != checkpoint.staging_multipart.is_some() {
+        return Ok(None);
+    }
+    let cleanup_started = sqlx::query_scalar::<_, bool>(
+        "WITH retained AS (
+             SELECT object_store, object_bucket, object_key, object_version_id
+             FROM result_inventory_members
+             WHERE attempt_id = $1
+             ORDER BY path COLLATE \"C\", member_id COLLATE \"C\"
+             LIMIT $2
+         )
+         SELECT EXISTS (
+             SELECT 1
+             FROM attempt_staging_cleanup c
+             JOIN retained r
+               ON r.object_store = c.target_store
+              AND r.object_bucket = c.bucket
+              AND r.object_key = c.object_key
+              AND r.object_version_id IS NOT DISTINCT FROM c.version_id
+             WHERE c.attempt_id = $1
+               AND c.kind = 'object'
+               AND c.state <> 'pending'
+         )",
+    )
+    .bind(owner)
+    .bind(validated_members)
+    .fetch_one(tx.as_mut())
+    .await?;
+    if cleanup_started {
+        return Ok(None);
+    }
+    if let Some(multipart) = &checkpoint.staging_multipart {
+        let available = sqlx::query_scalar::<_, bool>(
+            "SELECT EXISTS (
+                 SELECT 1 FROM attempt_staging_cleanup
+                 WHERE attempt_id = $1 AND kind = 'multipart'
+                   AND target_store = $2 AND bucket = $3
+                   AND object_key = $4 AND upload_id = $5
+                   AND state = 'pending'
+             )",
+        )
+        .bind(owner)
+        .bind(&multipart.target_store)
+        .bind(&multipart.bucket)
+        .bind(&multipart.key)
+        .bind(&multipart.upload_id)
+        .fetch_one(tx.as_mut())
+        .await?;
+        if !available {
+            return Ok(None);
+        }
+    }
+    Ok(Some(CheckpointStage {
+        owner_attempt_id: OpaqueId::parse(owner).map_err(DurableError::effect)?,
+        staging_origin_attempt_id: OpaqueId::parse(staging_origin).map_err(DurableError::effect)?,
+        validated_bytes: checkpoint.validated_bytes,
+        validated_members: checkpoint.validated_members,
+        inventory_bytes: u64::try_from(retained_bytes).map_err(DurableError::effect)?,
+        staging_multipart: checkpoint.staging_multipart.clone(),
+    }))
+}
+
+async fn transfer_checkpoint_stage(
+    tx: &mut remora_durable::PgTx<'_>,
+    stage: &CheckpointStage,
+    replacement: &OpaqueId,
+) -> Result<(), DurableError> {
+    let retained_members = i64::try_from(stage.validated_members).map_err(DurableError::effect)?;
+    let retained_bytes = i64::try_from(stage.inventory_bytes).map_err(DurableError::effect)?;
+    let now = OffsetDateTime::now_utc();
+    sqlx::query(
+        "WITH ranked AS (
+             SELECT *, ROW_NUMBER() OVER (
+                 ORDER BY path COLLATE \"C\", member_id COLLATE \"C\"
+             ) AS ordinal
+             FROM result_inventory_members
+             WHERE attempt_id = $1
+         )
+         INSERT INTO attempt_staging_cleanup (
+             attempt_id, kind, target_store, bucket, object_key, version_id, created_at
+         )
+         SELECT $1, 'object', object_store, object_bucket, object_key,
+                object_version_id, $3
+         FROM ranked
+         WHERE ordinal > $2
+         ON CONFLICT DO NOTHING",
+    )
+    .bind(stage.owner_attempt_id.as_str())
+    .bind(retained_members)
+    .bind(now)
+    .execute(tx.as_mut())
+    .await?;
+    sqlx::query(
+        "WITH retained AS (
+             SELECT object_store, object_bucket, object_key, object_version_id
+             FROM result_inventory_members
+             WHERE attempt_id = $1
+             ORDER BY path COLLATE \"C\", member_id COLLATE \"C\"
+             LIMIT $3
+         )
+         UPDATE attempt_staging_cleanup c
+         SET attempt_id = $2
+         FROM retained r
+         WHERE c.attempt_id = $1
+           AND c.kind = 'object'
+           AND c.state = 'pending'
+           AND r.object_store = c.target_store
+           AND r.object_bucket = c.bucket
+           AND r.object_key = c.object_key
+           AND r.object_version_id IS NOT DISTINCT FROM c.version_id",
+    )
+    .bind(stage.owner_attempt_id.as_str())
+    .bind(replacement.as_str())
+    .bind(retained_members)
+    .execute(tx.as_mut())
+    .await?;
+    if let Some(multipart) = &stage.staging_multipart {
+        let changed = sqlx::query(
+            "UPDATE attempt_staging_cleanup
+             SET attempt_id = $2
+             WHERE attempt_id = $1 AND state = 'pending'
+               AND (
+                   (kind = 'multipart' AND target_store = $3 AND bucket = $4
+                    AND object_key = $5 AND upload_id = $6)
+                   OR (kind = 'write_intent' AND target_store = $3 AND bucket = $4
+                       AND object_key = $5)
+               )",
+        )
+        .bind(stage.owner_attempt_id.as_str())
+        .bind(replacement.as_str())
+        .bind(&multipart.target_store)
+        .bind(&multipart.bucket)
+        .bind(&multipart.key)
+        .bind(&multipart.upload_id)
+        .execute(tx.as_mut())
+        .await?;
+        if changed.rows_affected() == 0 {
+            return Err(DurableError::effect(
+                "checkpoint multipart ownership transfer failed",
+            ));
+        }
+    }
+    sqlx::query(
+        "WITH extras AS (
+             SELECT attempt_id, member_id
+             FROM (
+                 SELECT attempt_id, member_id, ROW_NUMBER() OVER (
+                     ORDER BY path COLLATE \"C\", member_id COLLATE \"C\"
+                 ) AS ordinal
+                 FROM result_inventory_members
+                 WHERE attempt_id = $1
+             ) ranked
+             WHERE ordinal > $2
+         )
+         DELETE FROM result_inventory_members m
+         USING extras
+         WHERE m.attempt_id = extras.attempt_id
+           AND m.member_id = extras.member_id",
+    )
+    .bind(stage.owner_attempt_id.as_str())
+    .bind(retained_members)
+    .execute(tx.as_mut())
+    .await?;
+    let changed = sqlx::query(
+        "UPDATE result_inventories
+         SET attempt_id = $2, member_count = $3, total_bytes = $4
+         WHERE attempt_id = $1 AND state = 'open'",
+    )
+    .bind(stage.owner_attempt_id.as_str())
+    .bind(replacement.as_str())
+    .bind(retained_members)
+    .bind(retained_bytes)
+    .execute(tx.as_mut())
+    .await?;
+    if changed.rows_affected() != 1 {
+        return Err(DurableError::effect(
+            "checkpoint staging inventory transfer failed",
+        ));
+    }
+    Ok(())
+}
+
 async fn load_execution_envelope_record(
     tx: &mut remora_durable::PgTx<'_>,
     job_id: &OpaqueId,
     attempt_id: &OpaqueId,
     fence: u64,
     worker: &OpaqueId,
+    staging_origin_attempt_id: &OpaqueId,
 ) -> Result<ExecutionEnvelopeRecord, DurableError> {
     type Row = (
         Json<serde_json::Value>,
@@ -804,6 +1125,7 @@ async fn load_execution_envelope_record(
         target_store: row.2,
         target_bucket: row.3,
         target_prefix: row.4.0,
+        staging_origin_attempt_id: staging_origin_attempt_id.clone(),
         execution_grant_id: grant_id,
         checkpoint: row.9.map(|value| value.0),
     })

crates/remora-attempt/src/types.rs

@@ -157,6 +157,7 @@ pub struct ExecutionEnvelopeRecord {
     pub target_store: String,
     pub target_bucket: String,
     pub target_prefix: Vec<String>,
+    pub staging_origin_attempt_id: OpaqueId,
     pub execution_grant_id: OpaqueId,
     pub checkpoint: Option<crate::CheckpointProjection>,
 }

crates/remora-attempt/tests/mutations.rs

@@ -18,7 +18,7 @@ use remora_domain::{
 use remora_durable::{MutationDisposition, PoolConfig, connect, migrate as durable_migrate};
 use remora_identity::{DefaultTarget, Principal};
 use remora_result::{PublicationInput, migrate as result_migrate};
-use remora_target::{ContentSha256, ObjectIdentity, StoredObject, attempt_manifest_key};
+use remora_target::{ContentSha256, ObjectIdentity, StoredObject, attempt_manifest_key_in};
 use sqlx::PgPool;
 
 struct Harness {
@@ -168,7 +168,7 @@ impl Harness {
     ) -> PublicationInput {
         let manifest_id = unique_id("manifest");
         let sha256 = format!("sha256:{}", "11".repeat(32));
-        let key = attempt_manifest_key(attempt_id, &manifest_id);
+        let key = attempt_manifest_key_in("users/alice", attempt_id, &manifest_id);
         let input = PublicationInput {
             attempt_id: attempt_id.clone(),
             manifest_id: manifest_id.clone(),
@@ -190,9 +190,10 @@ impl Harness {
             "INSERT INTO result_inventories (
                 attempt_id, state, manifest_schema_version, manifest_id, manifest_store,
                 manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
-                manifest_size, member_count, total_bytes
+                manifest_size, member_count, total_bytes,
+                staging_origin_attempt_id, target_prefix
              ) VALUES ($1, 'sealed', 1, $2, 'primary', 'remora-results', $3,
-                       'version-1', $4, 128, $5, $6)",
+                       'version-1', $4, 128, $5, $6, $1, 'users/alice')",
         )
         .bind(attempt_id.as_str())
         .bind(manifest_id.as_str())
@@ -283,6 +284,7 @@ async fn resolved_source_checkpoint_and_progress_are_fenced_and_replayable() {
                     resolved_source_id: resolved.id.clone(),
                     validated_bytes: 0,
                     validated_members: 0,
+                    staging_multipart: None,
                 },
             )
             .await
@@ -295,6 +297,7 @@ async fn resolved_source_checkpoint_and_progress_are_fenced_and_replayable() {
         resolved_source_id: resolved.id.clone(),
         validated_bytes: 1024,
         validated_members: 1,
+        staging_multipart: None,
     };
     let mut wrong_source = checkpoint.clone();
     wrong_source.resolved_source_id = unique_id("resolved");
@@ -539,6 +542,7 @@ async fn failure_disposition_retries_with_budget_and_becomes_terminal_when_exhau
         resolved_source_id: retry_source.id.clone(),
         validated_bytes: 512,
         validated_members: 1,
+        staging_multipart: None,
     };
     assert_rejected(
         service
@@ -555,6 +559,40 @@ async fn failure_disposition_retries_with_budget_and_becomes_terminal_when_exhau
         .persist_checkpoint(&first, &unique_id("checkpointop"), retry_checkpoint)
         .await
         .unwrap();
+    sqlx::query(
+        "INSERT INTO result_inventories (
+            attempt_id, staging_origin_attempt_id, target_prefix
+         ) VALUES ($1, $1, 'users/alice')",
+    )
+    .bind(first.attempt_id.as_str())
+    .execute(&harness.pool)
+    .await
+    .unwrap();
+    sqlx::query(
+        "INSERT INTO result_inventory_members (
+            attempt_id, member_id, path, size, object_store, object_bucket,
+            object_key, object_sha256
+         ) VALUES ($1, $2, 'retry.bin', 512, 'target', 'bucket', $3, $4)",
+    )
+    .bind(first.attempt_id.as_str())
+    .bind(unique_id("member").as_str())
+    .bind(format!(
+        "users/alice/_remora/a/{}/o/retry/retry.bin",
+        first.attempt_id.as_str()
+    ))
+    .bind(format!("sha256:{}", "0".repeat(64)))
+    .execute(&harness.pool)
+    .await
+    .unwrap();
+    sqlx::query(
+        "UPDATE result_inventories
+         SET member_count = 1, total_bytes = 512
+         WHERE attempt_id = $1",
+    )
+    .bind(first.attempt_id.as_str())
+    .execute(&harness.pool)
+    .await
+    .unwrap();
     assert!(matches!(
         service
             .publish_failure(
@@ -610,6 +648,20 @@ async fn failure_disposition_retries_with_budget_and_becomes_terminal_when_exhau
             .await
             .unwrap();
     assert_eq!(initial_progress, 512);
+    let staging_origin: String =
+        sqlx::query_scalar("SELECT staging_origin_attempt_id FROM attempts WHERE attempt_id = $1")
+            .bind(second.attempt_id.as_str())
+            .fetch_one(&harness.pool)
+            .await
+            .unwrap();
+    assert_eq!(staging_origin, first.attempt_id.as_str());
+    let inventory_owner: String =
+        sqlx::query_scalar("SELECT attempt_id FROM result_inventories WHERE attempt_id = $1")
+            .bind(second.attempt_id.as_str())
+            .fetch_one(&harness.pool)
+            .await
+            .unwrap();
+    assert_eq!(inventory_owner, second.attempt_id.as_str());
     service
         .publish_failure(
             &second,
@@ -760,7 +812,7 @@ async fn publication_from_database(pool: &PgPool, attempt_id: &OpaqueId) -> Publ
             identity: ObjectIdentity {
                 target_store_id: row.1,
                 bucket: row.2,
-                key: attempt_manifest_key(attempt_id, &manifest_id),
+                key: attempt_manifest_key_in("users/alice", attempt_id, &manifest_id),
                 version_id: row.3,
             },
             content_sha256: ContentSha256::parse(row.4).unwrap(),

crates/remora-attempt/tests/recovery.rs

@@ -8,7 +8,7 @@ use remora_admission::{
 use remora_attempt::{
     ActivationOutcome, AttemptConfig, AttemptService, CancelOutcome, CleanupCompletion,
     CleanupIdentity, CooperativeStopOutcome, RecoveryConfig, StagingMultipart, StagingObject,
-    migrate as attempt_migrate,
+    StagingWriteIntent, migrate as attempt_migrate,
 };
 use remora_contracts::{HttpSourceSubmissionDto, JobSubmissionDto, SourceSubmissionDto};
 use remora_coordination::{CoordinationStore, StoreConfig};
@@ -21,7 +21,7 @@ use remora_identity::{DefaultTarget, Principal};
 use remora_result::{PublicationInput, migrate as result_migrate};
 use remora_target::{
     CreateObjectRequest, MultipartIdentity, ObjectIdentity, StoredObject, TargetStore,
-    TargetStoreConfig, attempt_manifest_key, attempt_object_key,
+    TargetStoreConfig, attempt_manifest_key_in, attempt_object_key_in,
 };
 use sqlx::PgPool;
 use tokio::io::copy_bidirectional;
@@ -264,8 +264,12 @@ impl Harness {
             "INSERT INTO result_inventories (
                 attempt_id, state, manifest_schema_version, manifest_id, manifest_store,
                 manifest_bucket, manifest_key, manifest_version_id, manifest_sha256,
-                manifest_size, member_count, total_bytes
+                manifest_size, member_count, total_bytes,
-             ) VALUES ($1, 'sealed', 1, $2, $3, $4, $5, $6, $7, $8, 0, 0)",
+                staging_origin_attempt_id, target_prefix
+             ) VALUES (
+                $1, 'sealed', 1, $2, $3, $4, $5, $6, $7, $8, 0, 0,
+                $1, 'users/recovery'
+             )",
         )
         .bind(authority.attempt_id.as_str())
         .bind(manifest_id.as_str())
@@ -765,7 +769,12 @@ async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner()
 
     let loser_job = harness.job().await;
     let loser = harness.executing(&service, &loser_job).await;
-    let loser_key = attempt_object_key(&loser.attempt_id, &unique_id("object"), None);
+    let loser_key = attempt_object_key_in(
+        "users/recovery",
+        &loser.attempt_id,
+        &unique_id("object"),
+        None,
+    );
     let loser_object = harness
         .target
         .put_create_only(
@@ -800,7 +809,12 @@ async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner()
             .await
             .is_err()
     );
-    let multipart_key = attempt_object_key(&loser.attempt_id, &unique_id("multipart"), None);
+    let multipart_key = attempt_object_key_in(
+        "users/recovery",
+        &loser.attempt_id,
+        &unique_id("multipart"),
+        None,
+    );
     let multipart = harness
         .target
         .begin_multipart(&CreateObjectRequest::parse(multipart_key.clone()).unwrap())
@@ -830,7 +844,8 @@ async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner()
 
     let winner_job = harness.job().await;
     let winner = harness.executing(&service, &winner_job).await;
-    let winner_key = attempt_manifest_key(&winner.attempt_id, &unique_id("manifest"));
+    let winner_key =
+        attempt_manifest_key_in("users/recovery", &winner.attempt_id, &unique_id("manifest"));
     let winner_object = harness
         .target
         .put_create_only(
@@ -935,7 +950,12 @@ async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner()
                 &StagingObject {
                     target_store: harness.target_config.target_store_id.clone(),
                     bucket: harness.target_config.bucket.clone(),
-                    key: attempt_object_key(&winner.attempt_id, &unique_id("wrong"), None),
+                    key: attempt_object_key_in(
+                        "users/recovery",
+                        &winner.attempt_id,
+                        &unique_id("wrong"),
+                        None,
+                    ),
                     version_id: None,
                 },
             )
@@ -944,11 +964,201 @@ async fn cleanup_uses_exact_durable_identities_retries_and_excludes_the_winner()
     );
 }
 
+#[tokio::test]
+async fn cleanup_reconciles_accepted_writes_from_durable_key_intents() {
+    let Some(harness) = Harness::new("write-intent").await else {
+        return;
+    };
+    let service = harness.service(1);
+    let config = recovery_config();
+
+    let multipart_job = harness.job().await;
+    let multipart_attempt = harness.executing(&service, &multipart_job).await;
+    let multipart_key = attempt_object_key_in(
+        "users/recovery",
+        &multipart_attempt.attempt_id,
+        &unique_id("multipart"),
+        None,
+    );
+    service
+        .register_staging_write_intent_operation(
+            &multipart_attempt,
+            &unique_id("intent"),
+            StagingWriteIntent {
+                target_store: harness.target_config.target_store_id.clone(),
+                bucket: harness.target_config.bucket.clone(),
+                key: multipart_key.clone(),
+            },
+        )
+        .await
+        .unwrap();
+    let accepted_multipart = harness
+        .target
+        .begin_multipart(&CreateObjectRequest::parse(multipart_key.clone()).unwrap())
+        .await
+        .unwrap();
+    let accepted_upload_id = accepted_multipart.identity().upload_id.clone();
+    service
+        .publish_failure(
+            &multipart_attempt,
+            &unique_id("failure"),
+            RuntimeFailureClass::Authorization,
+        )
+        .await
+        .unwrap();
+    let intent_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
+    assert!(matches!(
+        intent_claim.identity,
+        CleanupIdentity::WriteIntent(_)
+    ));
+    let discovered = harness
+        .target
+        .discover_staging_write(&multipart_key)
+        .await
+        .unwrap();
+    assert!(discovered.objects.is_empty());
+    assert_eq!(discovered.multiparts.len(), 1);
+    assert_eq!(discovered.multiparts[0].upload_id, accepted_upload_id);
+    assert_eq!(
+        service
+            .reconcile_write_intent(
+                &intent_claim,
+                Vec::new(),
+                discovered
+                    .multiparts
+                    .into_iter()
+                    .map(|identity| StagingMultipart {
+                        target_store: identity.target_store_id,
+                        bucket: identity.bucket,
+                        key: identity.key,
+                        upload_id: identity.upload_id,
+                    })
+                    .collect(),
+            )
+            .await
+            .unwrap(),
+        CleanupCompletion::Completed
+    );
+    let multipart_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
+    assert!(matches!(
+        multipart_claim.identity,
+        CleanupIdentity::Multipart(_)
+    ));
+    execute_cleanup(&harness.target, &multipart_claim)
+        .await
+        .unwrap();
+    service.complete_cleanup(&multipart_claim).await.unwrap();
+    assert!(
+        harness
+            .target
+            .discover_staging_write(&multipart_key)
+            .await
+            .unwrap()
+            .multiparts
+            .is_empty()
+    );
+
+    let object_job = harness.job().await;
+    let object_attempt = harness.executing(&service, &object_job).await;
+    let object_key = attempt_object_key_in(
+        "users/recovery",
+        &object_attempt.attempt_id,
+        &unique_id("object"),
+        None,
+    );
+    service
+        .register_staging_write_intent_operation(
+            &object_attempt,
+            &unique_id("intent"),
+            StagingWriteIntent {
+                target_store: harness.target_config.target_store_id.clone(),
+                bucket: harness.target_config.bucket.clone(),
+                key: object_key.clone(),
+            },
+        )
+        .await
+        .unwrap();
+    let accepted_object = harness
+        .target
+        .put_create_only(
+            &CreateObjectRequest::parse(object_key.clone()).unwrap(),
+            b"accepted-before-ledger".as_slice(),
+        )
+        .await
+        .unwrap();
+    service
+        .publish_failure(
+            &object_attempt,
+            &unique_id("failure"),
+            RuntimeFailureClass::Authorization,
+        )
+        .await
+        .unwrap();
+    let intent_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
+    let discovered = harness
+        .target
+        .discover_staging_write(&object_key)
+        .await
+        .unwrap();
+    let [discovered_object] = discovered.objects.as_slice() else {
+        panic!("expected one accepted object");
+    };
+    assert_eq!(
+        discovered_object.target_store_id,
+        accepted_object.identity.target_store_id
+    );
+    assert_eq!(discovered_object.bucket, accepted_object.identity.bucket);
+    assert_eq!(discovered_object.key, accepted_object.identity.key);
+    if let (Some(discovered_version), Some(accepted_version)) = (
+        discovered_object.version_id.as_deref(),
+        accepted_object.identity.version_id.as_deref(),
+    ) {
+        assert_eq!(discovered_version, accepted_version);
+    }
+    assert!(discovered.multiparts.is_empty());
+    assert_eq!(
+        service
+            .reconcile_write_intent(
+                &intent_claim,
+                discovered
+                    .objects
+                    .into_iter()
+                    .map(|identity| StagingObject {
+                        target_store: identity.target_store_id,
+                        bucket: identity.bucket,
+                        key: identity.key,
+                        version_id: identity.version_id,
+                    })
+                    .collect(),
+                Vec::new(),
+            )
+            .await
+            .unwrap(),
+        CleanupCompletion::Completed
+    );
+    let object_claim = service.claim_cleanup(&config).await.unwrap().unwrap();
+    assert!(matches!(object_claim.identity, CleanupIdentity::Object(_)));
+    execute_cleanup(&harness.target, &object_claim)
+        .await
+        .unwrap();
+    service.complete_cleanup(&object_claim).await.unwrap();
+    assert!(
+        harness
+            .target
+            .head_exact(&accepted_object.identity)
+            .await
+            .is_err()
+    );
+}
+
 async fn execute_cleanup(
     target: &TargetStore,
     claim: &remora_attempt::CleanupClaim,
 ) -> Result<(), remora_target::TargetError> {
     match &claim.identity {
+        CleanupIdentity::WriteIntent(_) => Err(remora_target::TargetError::internal(
+            "write intents must be reconciled before exact cleanup",
+        )),
         CleanupIdentity::Object(object) => {
             target
                 .delete_exact(&ObjectIdentity {

crates/remora-contracts/src/internal.rs

@@ -71,6 +71,78 @@ pub struct AttemptActivateRequestDto {
     pub worker_incarnation: OpaqueId,
 }
 
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct AttemptStopRequestDto {
+    pub operation_id: OpaqueId,
+    pub fence: PositiveIntegerDto,
+    pub worker_incarnation: OpaqueId,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct AttemptStagingObjectRequestDto {
+    pub operation_id: OpaqueId,
+    pub fence: PositiveIntegerDto,
+    pub worker_incarnation: OpaqueId,
+    pub object: StagingObjectIdentityDto,
+    #[serde(
+        default,
+        deserialize_with = "crate::dto::optional_non_null",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub completed_upload_id: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct StagingObjectIdentityDto {
+    pub store: String,
+    pub bucket: String,
+    pub key: String,
+    #[serde(
+        default,
+        deserialize_with = "crate::dto::optional_non_null",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub version_id: Option<String>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct AttemptStagingWriteIntentRequestDto {
+    pub operation_id: OpaqueId,
+    pub fence: PositiveIntegerDto,
+    pub worker_incarnation: OpaqueId,
+    pub intent: StagingWriteIntentDto,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct StagingWriteIntentDto {
+    pub store: String,
+    pub bucket: String,
+    pub key: String,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct AttemptStagingMultipartRequestDto {
+    pub operation_id: OpaqueId,
+    pub fence: PositiveIntegerDto,
+    pub worker_incarnation: OpaqueId,
+    pub multipart: StagingMultipartIdentityDto,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields)]
+pub struct StagingMultipartIdentityDto {
+    pub store: String,
+    pub bucket: String,
+    pub key: String,
+    pub upload_id: String,
+}
+
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct ResolvedSourceDescriptorRefDto {
@@ -160,6 +232,12 @@ pub struct AttemptCheckpointRequestDto {
     pub resolved_source_id: OpaqueId,
     pub validated_bytes: SafeIntegerDto,
     pub validated_members: SafeIntegerDto,
+    #[serde(
+        default,
+        deserialize_with = "crate::dto::optional_non_null",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub staging_multipart: Option<StagingMultipartIdentityDto>,
 }
 
 #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]

crates/remora-contracts/src/lib.rs

@@ -30,9 +30,11 @@ pub use events::{
 pub use internal::{
     ActivationResponseDto, AttemptActivateRequestDto, AttemptCheckpointRequestDto,
     AttemptMutationDto, AttemptProgressRequestDto, AttemptRenewRequestDto,
-    AttemptResolvedSourceRequestDto, CheckpointDescriptorKindDto, CheckpointDescriptorRefDto,
+    AttemptResolvedSourceRequestDto, AttemptStagingMultipartRequestDto,
-    ClaimRequestDto, ClaimResponseDto, ExecutionEnvelopeDto, ResolvedSourceDescriptorKindDto,
+    AttemptStagingObjectRequestDto, AttemptStagingWriteIntentRequestDto, AttemptStopRequestDto,
-    ResolvedSourceDescriptorRefDto,
+    CheckpointDescriptorKindDto, CheckpointDescriptorRefDto, ClaimRequestDto, ClaimResponseDto,
+    ExecutionEnvelopeDto, ResolvedSourceDescriptorKindDto, ResolvedSourceDescriptorRefDto,
+    StagingMultipartIdentityDto, StagingObjectIdentityDto, StagingWriteIntentDto,
 };
 pub use public::{
     ApiErrorCodeDto, ApiErrorDto, ApiErrorFieldDto, AttemptPageDto, AttemptRecordDto,

crates/remora-contracts/tests/schema_contract.rs

@@ -3,8 +3,9 @@ use std::{collections::BTreeSet, fs};
 use remora_contracts::{
     ActivationResponseDto, AttemptActivateRequestDto, AttemptCheckpointRequestDto,
     AttemptOutcomeRequestDto, AttemptProgressRequestDto, AttemptRenewRequestDto,
-    AttemptResolvedSourceRequestDto, ClaimRequestDto, ClaimResponseDto, JobPhaseProjectionDto,
+    AttemptResolvedSourceRequestDto, AttemptStagingMultipartRequestDto,
-    LeaseProjectionDto,
+    AttemptStagingObjectRequestDto, AttemptStagingWriteIntentRequestDto, AttemptStopRequestDto,
+    ClaimRequestDto, ClaimResponseDto, JobPhaseProjectionDto, LeaseProjectionDto,
 };
 use serde::{Serialize, de::DeserializeOwned};
 use serde_json::Value;
@@ -96,7 +97,7 @@ fn worker_routes_and_dtos_conform_to_the_mutable_openapi_contract() {
         "staging_target": {
             "store": "primary",
             "bucket": "remora-results",
-            "prefix": "_remora/a/attempt_0001/"
+            "prefix": "users/alice/_remora/a/attempt_0001/"
         },
         "execution_grant_ref": "grant_0001",
         "checkpoint_ref": null
@@ -135,6 +136,58 @@ fn worker_routes_and_dtos_conform_to_the_mutable_openapi_contract() {
         "AttemptActivateRequest",
         mutation.clone(),
     );
+    assert_dto_and_schema::<AttemptStopRequestDto>(
+        &component_schema,
+        "AttemptStopRequest",
+        serde_json::json!({
+            "operation_id": "operation_0001",
+            "fence": 1,
+            "worker_incarnation": "worker_0001"
+        }),
+    );
+    assert_dto_and_schema::<AttemptStagingObjectRequestDto>(
+        &component_schema,
+        "AttemptStagingObjectRequest",
+        serde_json::json!({
+            "operation_id": "operation_0001",
+            "fence": 1,
+            "worker_incarnation": "worker_0001",
+            "object": {
+                "store": "primary",
+                "bucket": "remora-results",
+                "key": "users/alice/_remora/a/attempt_0001/o/object_0001"
+            }
+        }),
+    );
+    assert_dto_and_schema::<AttemptStagingWriteIntentRequestDto>(
+        &component_schema,
+        "AttemptStagingWriteIntentRequest",
+        serde_json::json!({
+            "operation_id": "operation_0001",
+            "fence": 1,
+            "worker_incarnation": "worker_0001",
+            "intent": {
+                "store": "primary",
+                "bucket": "remora-results",
+                "key": "users/alice/_remora/a/attempt_0001/o/object_0001"
+            }
+        }),
+    );
+    assert_dto_and_schema::<AttemptStagingMultipartRequestDto>(
+        &component_schema,
+        "AttemptStagingMultipartRequest",
+        serde_json::json!({
+            "operation_id": "operation_0001",
+            "fence": 1,
+            "worker_incarnation": "worker_0001",
+            "multipart": {
+                "store": "primary",
+                "bucket": "remora-results",
+                "key": "users/alice/_remora/a/attempt_0001/o/object_0001",
+                "upload_id": "upload_0001"
+            }
+        }),
+    );
     assert_dto_and_schema::<ActivationResponseDto>(
         &component_schema,
         "ActivationResponse",
@@ -228,6 +281,30 @@ fn worker_routes_and_dtos_conform_to_the_mutable_openapi_contract() {
             "200",
             Some("ActivationResponse"),
         ),
+        (
+            "/internal/v1/attempts/{attempt_id}/stop",
+            "AttemptStopRequest",
+            "204",
+            None,
+        ),
+        (
+            "/internal/v1/attempts/{attempt_id}/staging/write-intent",
+            "AttemptStagingWriteIntentRequest",
+            "204",
+            None,
+        ),
+        (
+            "/internal/v1/attempts/{attempt_id}/staging/object",
+            "AttemptStagingObjectRequest",
+            "204",
+            None,
+        ),
+        (
+            "/internal/v1/attempts/{attempt_id}/staging/multipart",
+            "AttemptStagingMultipartRequest",
+            "204",
+            None,
+        ),
         (
             "/internal/v1/attempts/{attempt_id}/resolved-source",
             "AttemptResolvedSourceRequest",

crates/remora-hf-source/src/source.rs

@@ -40,6 +40,26 @@ impl<R: Resolver> HfSource<R> {
         }
     }
 
+    pub fn new_with_root_certificates(
+        resolver: R,
+        policy: OutboundPolicy,
+        timing: RetryTiming,
+        roots: impl IntoIterator<Item = Vec<u8>>,
+    ) -> Result<Self, HfSourceError> {
+        let timing = timing.sanitized();
+        let mut client =
+            OutboundClient::builder(resolver, policy).connect_timeout(timing.connect_timeout);
+        for root in roots {
+            let root =
+                reqwest::Certificate::from_der(&root).map_err(|_| HfSourceError::InvalidRequest)?;
+            client = client.add_root_certificate(root);
+        }
+        Ok(Self {
+            client: client.build(),
+            timing,
+        })
+    }
+
     pub async fn resolve(
         &self,
         request: HfSourceRequest,