diff --git a/backend/alembic/versions/0060_submission_bundle_durable_intent.py b/backend/alembic/versions/0060_submission_bundle_durable_intent.py
new file mode 100644
index 00000000..87d24101
--- /dev/null
+++ b/backend/alembic/versions/0060_submission_bundle_durable_intent.py
@@ -0,0 +1,251 @@
+"""Install submission-bundle durable put intent.
+
+Revision ID: 0060_submission_bundle_intent
+Revises: 0059_policy_execution_claim
+Create Date: 2026-08-07
+"""
+
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0060_submission_bundle_intent"
+down_revision = "0059_policy_execution_claim"
+branch_labels = depends_on = None
+
+_UUID = (
+    r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-"
+    r"[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
+)
+_CURRENT_REQUEST_TYPES = "producer_request_type in ('guide', 'checker_output')"
+_SUBMISSION_REQUEST_TYPES = (
+    "producer_request_type in ('guide', 'checker_output', 'submission_bundle')"
+)
+_CURRENT_PRODUCER_IDENTITY = (
+    "((producer_request_type = 'guide' and producer_type = 'actor_profile' "
+    f"and producer_ref ~ '{_UUID}') or "
+    "(producer_request_type = 'checker_output' and producer_type = 'service_identity' "
+    "and producer_ref = 'workstream.artifact.checker_output'))"
+)
+_SUBMISSION_PRODUCER_IDENTITY = (
+    "((producer_request_type = 'guide' and producer_type = 'actor_profile' "
+    f"and producer_ref ~ '{_UUID}') or "
+    "(producer_request_type = 'checker_output' and producer_type = 'service_identity' "
+    "and producer_ref = 'workstream.artifact.checker_output') or "
+    "(producer_request_type = 'submission_bundle' and producer_type = 'actor_profile' "
+    f"and producer_ref ~ '{_UUID}'))"
+)
+_CURRENT_PRODUCER_REFERENCE = (
+    "(producer_request_type = 'guide' and guide_source_item_id is not null "
+    "and checker_run_id is null and task_id is null and logical_role is null) or "
+    "(producer_request_type = 'checker_output' and guide_source_item_id is null "
+    "and checker_run_id is not null and task_id is not null "
+    "and octet_length(logical_role) between 1 and 100)"
+)
+_SUBMISSION_PRODUCER_REFERENCE = (
+    f"{_CURRENT_PRODUCER_REFERENCE} or "
+    "(producer_request_type = 'submission_bundle' and guide_source_item_id is null "
+    "and checker_run_id is null and task_id is not null and logical_role is null)"
+)
+_CURRENT_RECEIPT_REFERENCE = (
+    "contract_version = 2 and put_attempt_id is not null and "
+    "((guide_source_item_id is not null)::int + "
+    "(checker_run_id is not null)::int) = 1"
+)
+_SUBMISSION_RECEIPT_REFERENCE = (
+    "contract_version = 2 and put_attempt_id is not null and "
+    "((guide_source_item_id is not null and checker_run_id is null "
+    "and logical_role is null) or "
+    "(guide_source_item_id is null and checker_run_id is not null "
+    "and octet_length(logical_role) between 1 and 100) or "
+    "(guide_source_item_id is null and checker_run_id is null "
+    "and logical_role is null))"
+)
+
+
+def _replace_put_constraint(name: str, expression: str) -> None:
+    op.drop_constraint(name, "artifact_put_attempts", type_="check")
+    op.create_check_constraint(name, "artifact_put_attempts", expression)
+
+
+def upgrade() -> None:
+    _replace_put_constraint("producer_request_type", _SUBMISSION_REQUEST_TYPES)
+    _replace_put_constraint("producer_identity", _SUBMISSION_PRODUCER_IDENTITY)
+    _replace_put_constraint("producer_reference", _SUBMISSION_PRODUCER_REFERENCE)
+    op.drop_constraint(
+        "contract_producer_reference",
+        "artifact_operation_receipts",
+        type_="check",
+    )
+    op.create_check_constraint(
+        "contract_producer_reference",
+        "artifact_operation_receipts",
+        _SUBMISSION_RECEIPT_REFERENCE,
+    )
+    op.execute(
+        """
+        create function guard_artifact_receipt_producer_reference()
+        returns trigger language plpgsql as $$
+        declare request_type text;
+        begin
+          select producer_request_type into request_type
+          from artifact_put_attempts where id = new.put_attempt_id;
+          if request_type is null
+             or (request_type = 'guide' and not (
+                 new.guide_source_item_id is not null and new.checker_run_id is null
+                 and new.logical_role is null))
+             or (request_type = 'checker_output' and not (
+                 new.guide_source_item_id is null and new.checker_run_id is not null
+                 and octet_length(new.logical_role) between 1 and 100))
+             or (request_type = 'submission_bundle' and not (
+                 new.guide_source_item_id is null and new.checker_run_id is null
+                 and new.logical_role is null))
+          then
+            raise exception 'artifact receipt producer reference mismatch'
+              using errcode='23514';
+          end if;
+          return new;
+        end;
+        $$
+        """
+    )
+    op.execute(
+        "create trigger artifact_receipt_producer_reference "
+        "before insert or update of put_attempt_id, guide_source_item_id, "
+        "checker_run_id, logical_role on artifact_operation_receipts "
+        "for each row execute function guard_artifact_receipt_producer_reference()"
+    )
+    op.create_table(
+        "submission_bundle_durable_intents",
+        sa.Column("id", sa.String(36), primary_key=True),
+        sa.Column("pre_submit_evidence_set_id", sa.String(36), nullable=False),
+        sa.Column("put_attempt_id", sa.String(36), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.UniqueConstraint(
+            "pre_submit_evidence_set_id",
+            name="uq_submission_bundle_intent_evidence",
+        ),
+        sa.UniqueConstraint(
+            "put_attempt_id",
+            name="uq_submission_bundle_intent_put_attempt",
+        ),
+        sa.ForeignKeyConstraint(
+            ["pre_submit_evidence_set_id"],
+            ["pre_submit_evidence_sets.id"],
+            ondelete="RESTRICT",
+        ),
+        sa.ForeignKeyConstraint(
+            ["put_attempt_id"],
+            ["artifact_put_attempts.id"],
+            ondelete="RESTRICT",
+        ),
+    )
+    op.create_index(
+        "ix_submission_bundle_durable_intents_pre_submit_evidence_set_id",
+        "submission_bundle_durable_intents",
+        ["pre_submit_evidence_set_id"],
+    )
+    op.create_index(
+        "ix_submission_bundle_durable_intents_put_attempt_id",
+        "submission_bundle_durable_intents",
+        ["put_attempt_id"],
+    )
+    op.execute(
+        """
+        create function guard_submission_bundle_durable_intent_put_attempt()
+        returns trigger language plpgsql as $$
+        declare request_type text;
+        begin
+          select producer_request_type into request_type
+          from artifact_put_attempts
+          where id = new.put_attempt_id
+          for share;
+          if request_type is distinct from 'submission_bundle' then
+            raise exception 'submission bundle durable intent requires submission_bundle put attempt'
+              using errcode='23514';
+          end if;
+          return new;
+        end;
+        $$
+        """
+    )
+    op.execute(
+        "create trigger submission_bundle_durable_intent_put_attempt "
+        "before insert on submission_bundle_durable_intents "
+        "for each row execute function "
+        "guard_submission_bundle_durable_intent_put_attempt()"
+    )
+    op.execute(
+        """
+        create function guard_submission_bundle_durable_intents_immutable()
+        returns trigger language plpgsql as $$
+        begin
+          raise exception 'submission_bundle_durable_intents rows are immutable'
+            using errcode='55000';
+        end;
+        $$
+        """
+    )
+    op.execute(
+        "create trigger submission_bundle_durable_intents_immutable "
+        "before update or delete on submission_bundle_durable_intents "
+        "for each row execute function "
+        "guard_submission_bundle_durable_intents_immutable()"
+    )
+    op.execute(
+        "create trigger submission_bundle_durable_intents_no_truncate "
+        "before truncate on submission_bundle_durable_intents "
+        "for each statement execute function "
+        "guard_submission_bundle_durable_intents_immutable()"
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if bind.execute(sa.text("select count(*) from submission_bundle_durable_intents")).scalar_one():
+        raise RuntimeError("cannot remove populated submission-bundle durable intents")
+    op.execute("drop trigger artifact_receipt_producer_reference on artifact_operation_receipts")
+    op.execute("drop function guard_artifact_receipt_producer_reference()")
+    op.execute(
+        "drop trigger submission_bundle_durable_intents_no_truncate "
+        "on submission_bundle_durable_intents"
+    )
+    op.execute(
+        "drop trigger submission_bundle_durable_intents_immutable "
+        "on submission_bundle_durable_intents"
+    )
+    op.execute("drop function guard_submission_bundle_durable_intents_immutable()")
+    op.execute(
+        "drop trigger submission_bundle_durable_intent_put_attempt "
+        "on submission_bundle_durable_intents"
+    )
+    op.execute("drop function guard_submission_bundle_durable_intent_put_attempt()")
+    op.drop_index(
+        "ix_submission_bundle_durable_intents_put_attempt_id",
+        table_name="submission_bundle_durable_intents",
+    )
+    op.drop_index(
+        "ix_submission_bundle_durable_intents_pre_submit_evidence_set_id",
+        table_name="submission_bundle_durable_intents",
+    )
+    op.drop_table("submission_bundle_durable_intents")
+    op.drop_constraint(
+        "contract_producer_reference",
+        "artifact_operation_receipts",
+        type_="check",
+    )
+    op.create_check_constraint(
+        "contract_producer_reference",
+        "artifact_operation_receipts",
+        _CURRENT_RECEIPT_REFERENCE,
+    )
+    _replace_put_constraint("producer_reference", _CURRENT_PRODUCER_REFERENCE)
+    _replace_put_constraint("producer_identity", _CURRENT_PRODUCER_IDENTITY)
+    _replace_put_constraint("producer_request_type", _CURRENT_REQUEST_TYPES)
diff --git a/backend/app/db/models.py b/backend/app/db/models.py
index 40617af0..07e66842 100644
--- a/backend/app/db/models.py
+++ b/backend/app/db/models.py
@@ -22,6 +22,7 @@
     ArtifactRecoveryAttempt,
     PreSubmitEvidenceResult,
     PreSubmitEvidenceSet,
+    SubmissionBundleDurableIntent,
 )
 from app.modules.authorization.models import (  # noqa: F401
     AdminRoleGrant,
diff --git a/backend/app/modules/artifacts/models.py b/backend/app/modules/artifacts/models.py
index d36bc91a..b8755c3d 100644
--- a/backend/app/modules/artifacts/models.py
+++ b/backend/app/modules/artifacts/models.py
@@ -237,7 +237,8 @@ class PreSubmitEvidenceResult(Base):
     __table_args__ = (
         UniqueConstraint("evidence_set_id", "result_order", name="uq_pre_submit_result_order"),
         UniqueConstraint(
-            "evidence_set_id", "definition_id",
+            "evidence_set_id",
+            "definition_id",
             name="uq_pre_submit_result_definition",
         ),
         CheckConstraint("result_order >= 0", name="ck_pre_submit_result_order"),
@@ -894,7 +895,7 @@ class ArtifactPutAttempt(Base):
         ),
         UniqueConstraint("operation_identity", name="uq_artifact_put_attempt_operation"),
         CheckConstraint(
-            "producer_request_type in ('guide', 'checker_output')",
+            "producer_request_type in ('guide', 'checker_output', 'submission_bundle')",
             name="producer_request_type",
         ),
         CheckConstraint(
@@ -907,7 +908,11 @@ class ArtifactPutAttempt(Base):
             + UUID_CHECK.format(column="producer_ref")
             + ") or (producer_request_type = 'checker_output' "
             "and producer_type = 'service_identity' "
-            "and producer_ref = 'workstream.artifact.checker_output'))",
+            "and producer_ref = 'workstream.artifact.checker_output') or "
+            "(producer_request_type = 'submission_bundle' "
+            "and producer_type = 'actor_profile' and "
+            + UUID_CHECK.format(column="producer_ref")
+            + "))",
             name="producer_identity",
         ),
         CheckConstraint(SHA256_CHECK.format(column="sha256"), name="sha256_shape"),
@@ -969,7 +974,10 @@ class ArtifactPutAttempt(Base):
             "and logical_role is null) or "
             "(producer_request_type = 'checker_output' and guide_source_item_id is null "
             "and checker_run_id is not null and task_id is not null "
-            "and octet_length(logical_role) between 1 and 100)",
+            "and octet_length(logical_role) between 1 and 100) or "
+            "(producer_request_type = 'submission_bundle' "
+            "and guide_source_item_id is null and checker_run_id is null "
+            "and task_id is not null and logical_role is null)",
             name="producer_reference",
         ),
     )
@@ -1025,6 +1033,37 @@ class ArtifactPutAttempt(Base):
     )
 
 
+class SubmissionBundleDurableIntent(Base):
+    """Immutable join fencing one passing evidence set to one provider intent."""
+
+    __tablename__ = "submission_bundle_durable_intents"
+    __table_args__ = (
+        UniqueConstraint(
+            "pre_submit_evidence_set_id",
+            name="uq_submission_bundle_intent_evidence",
+        ),
+        UniqueConstraint(
+            "put_attempt_id",
+            name="uq_submission_bundle_intent_put_attempt",
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True)
+    pre_submit_evidence_set_id: Mapped[str] = mapped_column(
+        ForeignKey("pre_submit_evidence_sets.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    put_attempt_id: Mapped[str] = mapped_column(
+        ForeignKey("artifact_put_attempts.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
 class ArtifactPutAttemptCharge(Base):
     """Immutable link from one put attempt to every required scope charge."""
 
@@ -1102,8 +1141,12 @@ class ArtifactOperationReceipt(Base):
         CheckConstraint("attempt_number > 0", name="attempt_positive"),
         CheckConstraint(
             "contract_version = 2 and put_attempt_id is not null and "
-            "((guide_source_item_id is not null)::int + "
-            "(checker_run_id is not null)::int) = 1",
+            "((guide_source_item_id is not null and checker_run_id is null "
+            "and logical_role is null) or "
+            "(guide_source_item_id is null and checker_run_id is not null "
+            "and octet_length(logical_role) between 1 and 100) or "
+            "(guide_source_item_id is null and checker_run_id is null "
+            "and logical_role is null))",
             name="contract_producer_reference",
         ),
     )
diff --git a/backend/app/modules/artifacts/pre_submit_evidence.py b/backend/app/modules/artifacts/pre_submit_evidence.py
index 24701f7e..df96b856 100644
--- a/backend/app/modules/artifacts/pre_submit_evidence.py
+++ b/backend/app/modules/artifacts/pre_submit_evidence.py
@@ -223,12 +223,18 @@ def semantic_manifest_identity(semantic_manifest_sha256: str) -> UUID:
     )
 
 
+_PRE_SUBMIT_PASS_CAPABILITY_SEAL = object()
+
+
 class PreSubmitPassCapability:
     """Single-use process-local proof for immediate 04C continuation only."""
 
     __slots__ = (
         "_consumed",
+        "_binding",
         "_lock",
+        "_owner",
+        "_seal",
         "evidence_set_id",
         "prepared_generation_id",
         "predecessor_submission_id",
@@ -238,9 +244,16 @@ class PreSubmitPassCapability:
         "storage_scheme",
     )
 
-    def __init__(
-        self,
+    def __init__(self, *_: object, **__: object) -> None:
+        """Reject direct construction outside the evidence service."""
+        raise TypeError("PreSubmitPassCapability can only be created by pre-submit evidence")
+
+    @classmethod
+    def _from_evidence_service(
+        cls,
         *,
+        owner: PreSubmitEvidenceService,
+        binding: object,
         evidence_set_id: UUID,
         prepared_generation_id: UUID,
         predecessor_submission_id: UUID | None,
@@ -248,16 +261,24 @@ def __init__(
         archive_sha256: str,
         semantic_manifest_sha256: str,
         storage_scheme: str,
-    ) -> None:
-        self.evidence_set_id = evidence_set_id
-        self.prepared_generation_id = prepared_generation_id
-        self.predecessor_submission_id = predecessor_submission_id
-        self.effective_plan_sha256 = effective_plan_sha256
-        self.archive_sha256 = archive_sha256
-        self.semantic_manifest_sha256 = semantic_manifest_sha256
-        self.storage_scheme = storage_scheme
-        self._consumed = False
-        self._lock = threading.Lock()
+    ) -> PreSubmitPassCapability:
+        """Mint only from the service that persisted fresh passing evidence."""
+        if type(owner) is not PreSubmitEvidenceService or not owner._claims_pass_binding(binding):
+            raise TypeError("PreSubmitPassCapability can only be created by pre-submit evidence")
+        capability = object.__new__(cls)
+        capability._owner = owner
+        capability._binding = binding
+        capability.evidence_set_id = evidence_set_id
+        capability.prepared_generation_id = prepared_generation_id
+        capability.predecessor_submission_id = predecessor_submission_id
+        capability.effective_plan_sha256 = effective_plan_sha256
+        capability.archive_sha256 = archive_sha256
+        capability.semantic_manifest_sha256 = semantic_manifest_sha256
+        capability.storage_scheme = storage_scheme
+        capability._consumed = False
+        capability._lock = threading.Lock()
+        capability._seal = _PRE_SUBMIT_PASS_CAPABILITY_SEAL
+        return capability
 
     def consume(
         self,
@@ -271,18 +292,41 @@ def consume(
     ) -> UUID:
         """Consume once only when the immediate continuation facts still match."""
         with self._lock:
-            if self._consumed or (
-                prepared_generation_id != self.prepared_generation_id
-                or predecessor_submission_id != self.predecessor_submission_id
-                or effective_plan_sha256 != self.effective_plan_sha256
-                or archive_sha256 != self.archive_sha256
-                or semantic_manifest_sha256 != self.semantic_manifest_sha256
-                or storage_scheme != self.storage_scheme
+            if (
+                self._seal is not _PRE_SUBMIT_PASS_CAPABILITY_SEAL
+                or self._consumed
+                or (
+                    prepared_generation_id != self.prepared_generation_id
+                    or predecessor_submission_id != self.predecessor_submission_id
+                    or effective_plan_sha256 != self.effective_plan_sha256
+                    or archive_sha256 != self.archive_sha256
+                    or semantic_manifest_sha256 != self.semantic_manifest_sha256
+                    or storage_scheme != self.storage_scheme
+                )
             ):
                 raise PreSubmitEvidenceConflict("pre_submit_pass_capability_invalid")
+            if not self._owner._consume_pass_binding(self._binding):
+                raise PreSubmitEvidenceConflict("pre_submit_pass_capability_invalid")
             self._consumed = True
             return self.evidence_set_id
 
+    def _assert_live_prepared_custody(
+        self,
+        *,
+        prepared_generation_id: UUID,
+        archive_sha256: str,
+    ) -> None:
+        """Fail before handoff when this capability is spent or for other bytes."""
+        with self._lock:
+            if (
+                self._seal is not _PRE_SUBMIT_PASS_CAPABILITY_SEAL
+                or self._consumed
+                or not self._owner._claims_pass_binding(self._binding)
+                or prepared_generation_id != self.prepared_generation_id
+                or archive_sha256 != self.archive_sha256
+            ):
+                raise PreSubmitEvidenceConflict("pre_submit_pass_capability_invalid")
+
 
 class _PreSubmitEvidenceRepository:
     """Insert or replay one immutable evidence set in the caller transaction."""
@@ -431,6 +475,48 @@ class PreSubmitEvidenceService:
     def __init__(self, session: AsyncSession) -> None:
         self._session = session
         self._repository = _PreSubmitEvidenceRepository(session)
+        self._live_pass_bindings: set[object] = set()
+
+    def _claims_pass_binding(self, binding: object) -> bool:
+        """Recognize only an issuance registered by this live service instance."""
+        return binding in getattr(self, "_live_pass_bindings", set())
+
+    def _consume_pass_binding(self, binding: object) -> bool:
+        """Retire one process-local issuance binding exactly once."""
+        if binding not in self._live_pass_bindings:
+            return False
+        self._live_pass_bindings.remove(binding)
+        return True
+
+    def _mint_pass_capability(
+        self,
+        *,
+        evidence_set_id: UUID,
+        prepared_generation_id: UUID,
+        predecessor_submission_id: UUID | None,
+        effective_plan_sha256: str,
+        archive_sha256: str,
+        semantic_manifest_sha256: str,
+        storage_scheme: str,
+    ) -> PreSubmitPassCapability:
+        """Register one unguessable issuance after fresh passing persistence."""
+        binding = object()
+        self._live_pass_bindings.add(binding)
+        try:
+            return PreSubmitPassCapability._from_evidence_service(
+                owner=self,
+                binding=binding,
+                evidence_set_id=evidence_set_id,
+                prepared_generation_id=prepared_generation_id,
+                predecessor_submission_id=predecessor_submission_id,
+                effective_plan_sha256=effective_plan_sha256,
+                archive_sha256=archive_sha256,
+                semantic_manifest_sha256=semantic_manifest_sha256,
+                storage_scheme=storage_scheme,
+            )
+        except BaseException:
+            self._live_pass_bindings.discard(binding)
+            raise
 
     async def persist(
         self, request: PreSubmitEvidencePersistenceRequest
@@ -508,7 +594,7 @@ async def persist(
             execution=request.execution,
         )
         pass_capability = (
-            PreSubmitPassCapability(
+            self._mint_pass_capability(
                 evidence_set_id=evidence.evidence_set_id,
                 prepared_generation_id=request.prepared_generation_id,
                 predecessor_submission_id=request.predecessor_submission_id,
diff --git a/backend/app/modules/artifacts/repository.py b/backend/app/modules/artifacts/repository.py
index 95bdbeb6..4f0f60e9 100644
--- a/backend/app/modules/artifacts/repository.py
+++ b/backend/app/modules/artifacts/repository.py
@@ -17,6 +17,7 @@
     ArtifactBinding,
     ArtifactContent,
     ArtifactOperationReceipt,
+    PreSubmitEvidenceSet,
     ArtifactPutAttempt,
     ArtifactPutAttemptCharge,
     ArtifactPutObservationReceipt,
@@ -25,6 +26,7 @@
     ArtifactReplica,
     ArtifactRecoveryAttempt,
     ArtifactStorageNamespace,
+    SubmissionBundleDurableIntent,
 )
 from app.modules.checkers.models import CheckerRun
 from app.modules.projects.models import (
@@ -384,6 +386,46 @@ async def get_put_attempt_by_operation(
             )
         )
 
+    async def lock_pre_submit_evidence_set(
+        self, evidence_set_id: str
+    ) -> PreSubmitEvidenceSet | None:
+        """Lock one immutable passing evidence identity for durable continuation."""
+        return await self._session.scalar(
+            select(PreSubmitEvidenceSet)
+            .where(PreSubmitEvidenceSet.id == evidence_set_id)
+            .with_for_update(key_share=True)
+            .execution_options(populate_existing=True)
+        )
+
+    async def get_submission_bundle_intent_by_evidence(
+        self, evidence_set_id: str
+    ) -> SubmissionBundleDurableIntent | None:
+        """Load the one durable intent already created from an evidence set."""
+        return await self._session.scalar(
+            select(SubmissionBundleDurableIntent).where(
+                SubmissionBundleDurableIntent.pre_submit_evidence_set_id == evidence_set_id
+            )
+        )
+
+    async def lock_submission_bundle_intent(
+        self, intent_id: str
+    ) -> SubmissionBundleDurableIntent | None:
+        """Lock one exact durable intent selected for bounded source replay."""
+        return await self._session.scalar(
+            select(SubmissionBundleDurableIntent)
+            .where(SubmissionBundleDurableIntent.id == intent_id)
+            .with_for_update(key_share=True)
+            .execution_options(populate_existing=True)
+        )
+
+    async def add_submission_bundle_intent(
+        self, intent: SubmissionBundleDurableIntent
+    ) -> SubmissionBundleDurableIntent:
+        """Flush the immutable evidence-to-put-attempt fence."""
+        self._session.add(intent)
+        await self._session.flush()
+        return intent
+
     async def lock_put_attempt(self, attempt_id: str) -> ArtifactPutAttempt | None:
         """Lock one exact attempt and refresh its current fence."""
         return await self._session.scalar(
diff --git a/backend/app/modules/artifacts/schemas.py b/backend/app/modules/artifacts/schemas.py
index a6cba317..dbfdc7e8 100644
--- a/backend/app/modules/artifacts/schemas.py
+++ b/backend/app/modules/artifacts/schemas.py
@@ -38,8 +38,29 @@ class CheckerOutputArtifactAdmissionRequest:
     source: CommittedArtifactSource
 
 
+@final
+@dataclass(frozen=True, slots=True)
+class SubmissionBundleArtifactAdmissionRequest:
+    """One passing prepared bundle selected for durable provider intent."""
+
+    pre_submit_evidence_set_id: UUID
+    custody: object
+    replay_durable_intent_id: UUID | None
+
+    @property
+    def source(self) -> CommittedArtifactSource:
+        """Expose only the source sealed into process-local prepared custody."""
+        from app.modules.artifacts.submission_custody import SubmissionBundlePreparedCustody
+
+        if type(self.custody) is not SubmissionBundlePreparedCustody:
+            raise TypeError("submission bundle prepared custody is unavailable")
+        return self.custody.source
+
+
 ArtifactAdmissionRequest: TypeAlias = (
-    GuideArtifactAdmissionRequest | CheckerOutputArtifactAdmissionRequest
+    GuideArtifactAdmissionRequest
+    | CheckerOutputArtifactAdmissionRequest
+    | SubmissionBundleArtifactAdmissionRequest
 )
 
 
@@ -117,6 +138,39 @@ class GuideSourceReadAuthorityFacts:
     media_type: str
 
 
+@final
+@dataclass(frozen=True, slots=True)
+class SubmissionBundleDurableIntentAuthorityFacts:
+    """Final locked contributor and custody facts bound before durable intent."""
+
+    actor_profile_id: UUID
+    identity_link_id: UUID
+    project_id: UUID
+    task_id: UUID
+    assignment_id: UUID
+    predecessor_submission_id: UUID | None
+    predecessor_submission_version: int | None
+    pre_submit_evidence_set_id: UUID
+    prepared_generation_id: UUID
+    guide_id: UUID
+    guide_version: str
+    source_snapshot_id: UUID
+    source_snapshot_sha256: str
+    effective_policy_id: UUID
+    effective_policy_sha256: str
+    pre_submit_policy_id: UUID
+    pre_submit_policy_sha256: str
+    effective_plan_sha256: str
+    semantic_manifest_id: UUID
+    semantic_manifest_sha256: str
+    archive_sha256: str
+    archive_byte_count: int
+    media_type: str
+    storage_scheme: str
+    operation_identity: str
+    replay_durable_intent_id: UUID | None
+
+
 class ArtifactInternalResourceType(StrEnum):
     """Closed artifact-owned resource types for internal actions."""
 
diff --git a/backend/app/modules/artifacts/service.py b/backend/app/modules/artifacts/service.py
index a916cb5b..a61929a3 100644
--- a/backend/app/modules/artifacts/service.py
+++ b/backend/app/modules/artifacts/service.py
@@ -7,7 +7,7 @@
 import sys
 from collections.abc import AsyncIterable, Awaitable, Callable
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from datetime import datetime, timedelta
 from uuid import UUID, uuid4
 
@@ -50,6 +50,8 @@
     ArtifactStorageNamespace,
     ArtifactVerificationJob,
     ArtifactVerificationReceipt,
+    PreSubmitEvidenceSet,
+    SubmissionBundleDurableIntent,
 )
 from app.modules.artifacts.metrics import (
     ArtifactAdmissionMetrics,
@@ -80,6 +82,11 @@
     CheckerOutputArtifactAdmissionRequest,
     GuideArtifactAdmissionRequest,
     GuideArtifactIngestAuthorityFacts,
+    SubmissionBundleArtifactAdmissionRequest,
+    SubmissionBundleDurableIntentAuthorityFacts,
+)
+from app.modules.artifacts.submission_authorization import (
+    SubmissionBundlePreparedAuthorization,
 )
 from app.modules.artifacts.sources import CommittedArtifactSource, PreparedArtifact
 from app.modules.authorization.runtime import (
@@ -161,6 +168,7 @@ class _AdmissionFacts:
     guide_source_snapshot_id: str | None
     checker_run_id: str | None
     logical_role: str | None
+    pre_submit_evidence_set_id: str | None
     operation_identity: str
 
 
@@ -389,9 +397,7 @@ async def resume_committed_put(
         """Replay absent bytes or observe an otherwise ambiguous prior put."""
         async with self._session.begin():
             attempt = await self._repo.lock_put_attempt(str(attempt_id))
-            replay_required = (
-                attempt is not None and attempt.status == "absent_replay_required"
-            )
+            replay_required = attempt is not None and attempt.status == "absent_replay_required"
         if replay_required:
             return await self.execute_committed_put(attempt_id=attempt_id, source=source)
         status = await self.resolve_put_attempt(attempt_id)
@@ -1787,6 +1793,7 @@ async def admit(
         request: ArtifactAdmissionRequest,
         *,
         guide_prepared_authorization: GuideArtifactPreparedAuthorization | None = None,
+        submission_prepared_authorization: SubmissionBundlePreparedAuthorization | None = None,
         prepared_authorization: PreparedAuthorizationHandle | None = None,
         existing_transaction: bool = False,
     ) -> ArtifactAdmissionResult:
@@ -1797,10 +1804,6 @@ async def admit(
             self._session,
             existing=existing_transaction,
         ):
-            namespace = await _claim_and_validate_storage_namespace(
-                self._repo,
-                self._namespace,
-            )
             if type(request) is GuideArtifactAdmissionRequest:
                 if (
                     guide_prepared_authorization is None
@@ -1859,7 +1862,48 @@ async def admit(
                     )
                 except ValueError as exc:
                     raise ArtifactAdmissionRelationshipError(str(exc)) from exc
-            facts = await self._derive_admission_facts(request)
+            submission_facts: _AdmissionFacts | None = None
+            if type(request) is SubmissionBundleArtifactAdmissionRequest:
+                if (
+                    submission_prepared_authorization is None
+                    or type(prepared_authorization) is not PreparedAuthorizationHandle
+                ):
+                    raise ArtifactAuthorityDeniedError(
+                        "submission bundle durable preparation is unavailable"
+                    )
+                authority_facts, submission_facts = await self._submission_bundle_facts(request)
+                replay_attempt = await self._submission_bundle_replay_attempt(
+                    request,
+                    authority_facts,
+                )
+                if replay_attempt is not None:
+                    authority_facts = replace(
+                        authority_facts,
+                        operation_identity=replay_attempt.operation_identity,
+                    )
+                consumed_evidence_id = request.custody.pass_capability.consume(
+                    prepared_generation_id=authority_facts.prepared_generation_id,
+                    predecessor_submission_id=authority_facts.predecessor_submission_id,
+                    effective_plan_sha256=authority_facts.effective_plan_sha256,
+                    archive_sha256=authority_facts.archive_sha256,
+                    semantic_manifest_sha256=authority_facts.semantic_manifest_sha256,
+                    storage_scheme=authority_facts.storage_scheme,
+                )
+                if consumed_evidence_id != authority_facts.pre_submit_evidence_set_id:
+                    raise ArtifactAdmissionRelationshipError(
+                        "submission bundle pass capability evidence changed"
+                    )
+                await submission_prepared_authorization.consume(
+                    prepared_authorization=prepared_authorization,
+                    facts=authority_facts,
+                )
+                if replay_attempt is not None:
+                    return await self._result(replay_attempt, replayed=True)
+            namespace = await _claim_and_validate_storage_namespace(
+                self._repo,
+                self._namespace,
+            )
+            facts = submission_facts or await self._derive_admission_facts(request)
             scopes = self._derive_scopes(facts)
             request_digest = canonical_json_hash(
                 {
@@ -1872,6 +1916,7 @@ async def admit(
                     "guide_source_item_id": facts.guide_source_item_id,
                     "checker_run_id": facts.checker_run_id,
                     "logical_role": facts.logical_role,
+                    "pre_submit_evidence_set_id": facts.pre_submit_evidence_set_id,
                     "sha256": commitment.sha256,
                     "byte_count": commitment.byte_count,
                     "media_type": commitment.media_type,
@@ -1908,6 +1953,14 @@ async def admit(
                     counter.scope_type, counter.counted_bytes, counter.limit_bytes
                 )
             if replay is not None:
+                if facts.pre_submit_evidence_set_id is not None:
+                    intent = await self._repo.get_submission_bundle_intent_by_evidence(
+                        facts.pre_submit_evidence_set_id
+                    )
+                    if intent is None or intent.put_attempt_id != replay.id:
+                        raise ArtifactAdmissionConfigurationError(
+                            "submission bundle replay intent is incomplete"
+                        )
                 linked_charge_ids = await self._repo.list_put_attempt_charge_ids(replay.id)
                 reserved_charge_ids = tuple(sorted(charge.id for charge in charges))
                 if linked_charge_ids != reserved_charge_ids:
@@ -1950,6 +2003,14 @@ async def admit(
                 terminal_at=None,
             )
             await self._repo.add_put_attempt(attempt, charges)
+            if facts.pre_submit_evidence_set_id is not None:
+                await self._repo.add_submission_bundle_intent(
+                    SubmissionBundleDurableIntent(
+                        id=str(uuid4()),
+                        pre_submit_evidence_set_id=facts.pre_submit_evidence_set_id,
+                        put_attempt_id=attempt.id,
+                    )
+                )
             return await self._result(attempt, replayed=False)
 
     @staticmethod
@@ -1958,6 +2019,7 @@ def _validate_request_boundary(request: ArtifactAdmissionRequest) -> None:
         if type(request) not in {
             GuideArtifactAdmissionRequest,
             CheckerOutputArtifactAdmissionRequest,
+            SubmissionBundleArtifactAdmissionRequest,
         }:
             raise TypeError("invalid artifact admission request")
         if type(request.source) is not CommittedArtifactSource:
@@ -1975,6 +2037,8 @@ def _validate_request_boundary(request: ArtifactAdmissionRequest) -> None:
             ):
                 raise TypeError("guide artifact lineage claims are incomplete")
             return
+        if type(request) is SubmissionBundleArtifactAdmissionRequest:
+            return
         context = request.authorization_context
         if type(context) not in {HumanAuthorizationContext, ServiceAuthorizationContext}:
             raise TypeError("invalid artifact admission authorization context")
@@ -2020,6 +2084,7 @@ async def _guide_facts(self, request: GuideArtifactAdmissionRequest) -> _Admissi
             guide_source_snapshot_id=row.guide_source_snapshot_id,
             checker_run_id=None,
             logical_role=None,
+            pre_submit_evidence_set_id=None,
             operation_identity=operation_identity,
         )
 
@@ -2071,6 +2136,246 @@ async def _checker_output_facts(
             guide_source_snapshot_id=None,
             checker_run_id=checker_run_id,
             logical_role=logical_role,
+            pre_submit_evidence_set_id=None,
+            operation_identity=operation_identity,
+        )
+
+    async def _submission_bundle_replay_attempt(
+        self,
+        request: SubmissionBundleArtifactAdmissionRequest,
+        current: SubmissionBundleDurableIntentAuthorityFacts,
+    ) -> ArtifactPutAttempt | None:
+        """Reuse one committed intent when fresh checked custody exactly matches it."""
+        if request.replay_durable_intent_id is None:
+            return None
+        intent = await self._repo.lock_submission_bundle_intent(
+            str(request.replay_durable_intent_id)
+        )
+        if intent is None:
+            raise ArtifactAdmissionRelationshipError(
+                "submission bundle replay intent is unavailable"
+            )
+        prior = await self._repo.lock_pre_submit_evidence_set(intent.pre_submit_evidence_set_id)
+        fresh = await self._repo.lock_pre_submit_evidence_set(
+            str(current.pre_submit_evidence_set_id)
+        )
+        attempt = await self._repo.lock_put_attempt(intent.put_attempt_id)
+        if (
+            prior is None
+            or fresh is None
+            or attempt is None
+            or attempt.producer_request_type != "submission_bundle"
+            or attempt.status
+            not in {"prepared", "acknowledgement_unknown", "absent_replay_required"}
+            or attempt.sha256 != current.archive_sha256
+            or attempt.byte_count != current.archive_byte_count
+            or attempt.media_type != current.media_type
+            or self._submission_replay_lineage(prior) != self._submission_replay_lineage(fresh)
+            or (
+                prior.actor_profile_id,
+                prior.identity_link_id,
+                prior.project_id,
+                prior.task_id,
+                prior.assignment_id,
+                prior.predecessor_submission_id,
+                prior.predecessor_submission_version,
+                prior.guide_id,
+                prior.guide_version,
+                prior.source_snapshot_id,
+                prior.source_snapshot_sha256,
+                prior.locked_artifact_policy_sha256,
+                prior.locked_checker_policy_sha256,
+                prior.effective_plan_sha256,
+                prior.semantic_manifest_sha256,
+                prior.archive_sha256,
+                prior.archive_byte_count,
+                prior.storage_scheme,
+            )
+            != (
+                str(current.actor_profile_id),
+                str(current.identity_link_id),
+                str(current.project_id),
+                str(current.task_id),
+                str(current.assignment_id),
+                str(current.predecessor_submission_id)
+                if current.predecessor_submission_id is not None
+                else None,
+                current.predecessor_submission_version,
+                str(current.guide_id),
+                current.guide_version,
+                str(current.source_snapshot_id),
+                current.source_snapshot_sha256,
+                current.effective_policy_sha256,
+                current.pre_submit_policy_sha256,
+                current.effective_plan_sha256,
+                current.semantic_manifest_sha256,
+                current.archive_sha256,
+                current.archive_byte_count,
+                current.storage_scheme,
+            )
+        ):
+            raise ArtifactAdmissionRelationshipError(
+                "submission bundle replay intent does not match fresh custody"
+            )
+        return attempt
+
+    @staticmethod
+    def _submission_replay_lineage(evidence: PreSubmitEvidenceSet) -> tuple[object, ...]:
+        """Return every publication-relevant evidence fact except fresh issuance identity."""
+        return (
+            evidence.actor_profile_id,
+            evidence.identity_link_id,
+            evidence.project_id,
+            evidence.task_id,
+            evidence.assignment_id,
+            evidence.predecessor_submission_id,
+            evidence.predecessor_submission_version,
+            evidence.archive_sha256,
+            evidence.archive_byte_count,
+            evidence.semantic_manifest_id,
+            evidence.semantic_manifest_sha256,
+            evidence.guide_id,
+            evidence.guide_version,
+            evidence.source_snapshot_id,
+            evidence.source_snapshot_sha256,
+            evidence.locked_guide_sha256,
+            evidence.effective_policy_id,
+            evidence.locked_artifact_policy_sha256,
+            evidence.pre_submit_policy_id,
+            evidence.locked_checker_policy_sha256,
+            evidence.effective_plan_sha256,
+            evidence.catalogue_id,
+            evidence.catalogue_version,
+            evidence.catalogue_manifest_sha256,
+            evidence.storage_scheme,
+            evidence.terminal_status,
+            evidence.eligible,
+            evidence.result_count,
+            evidence.result_manifest_sha256,
+        )
+
+    async def _submission_bundle_facts(
+        self, request: SubmissionBundleArtifactAdmissionRequest
+    ) -> tuple[SubmissionBundleDurableIntentAuthorityFacts, _AdmissionFacts]:
+        """Lock exact passing evidence and current TASK-owned lineage."""
+        from app.modules.tasks.pre_submit_context import (
+            PreSubmitLockedContextInvalid,
+            load_locked_pre_submit_context,
+        )
+
+        from app.modules.artifacts.pre_submit_evidence import PreSubmitPassCapability
+        from app.modules.artifacts.submission_custody import SubmissionBundlePreparedCustody
+
+        if type(request.custody) is not SubmissionBundlePreparedCustody:
+            raise ArtifactAuthorityDeniedError("submission bundle prepared custody is unavailable")
+        pass_capability = request.custody.pass_capability
+        if type(pass_capability) is not PreSubmitPassCapability:
+            raise ArtifactAuthorityDeniedError("submission bundle pass capability is unavailable")
+        evidence = await self._repo.lock_pre_submit_evidence_set(
+            str(request.pre_submit_evidence_set_id)
+        )
+        commitment = request.source.commitment
+        if (
+            evidence is None
+            or evidence.terminal_status != "passed"
+            or not evidence.eligible
+            or evidence.prepared_generation_id != str(request.custody.prepared_generation_id)
+            or evidence.archive_sha256 != commitment.sha256
+            or evidence.archive_byte_count != commitment.byte_count
+            or commitment.media_type != "application/zip"
+            or evidence.predecessor_submission_id
+            != (
+                str(pass_capability.predecessor_submission_id)
+                if pass_capability.predecessor_submission_id is not None
+                else None
+            )
+            or evidence.effective_plan_sha256 != pass_capability.effective_plan_sha256
+            or evidence.semantic_manifest_sha256 != pass_capability.semantic_manifest_sha256
+            or evidence.storage_scheme != pass_capability.storage_scheme
+        ):
+            raise ArtifactAdmissionRelationshipError(
+                "submission bundle passing evidence is unavailable"
+            )
+        try:
+            locked = await load_locked_pre_submit_context(
+                self._session,
+                actor_profile_id=UUID(evidence.actor_profile_id),
+                identity_link_id=UUID(evidence.identity_link_id),
+                task_id=UUID(evidence.task_id),
+                assignment_id=UUID(evidence.assignment_id),
+                predecessor_submission_id=(
+                    UUID(evidence.predecessor_submission_id)
+                    if evidence.predecessor_submission_id is not None
+                    else None
+                ),
+                include_actor_identity_locks=False,
+            )
+        except PreSubmitLockedContextInvalid as exc:
+            raise ArtifactAdmissionRelationshipError(str(exc)) from exc
+        if (
+            locked.project_id != UUID(evidence.project_id)
+            or locked.predecessor_submission_version != evidence.predecessor_submission_version
+            or locked.guide_id != UUID(evidence.guide_id)
+            or locked.guide_version != evidence.guide_version
+            or locked.source_snapshot_id != UUID(evidence.source_snapshot_id)
+            or locked.source_snapshot_sha256 != evidence.source_snapshot_sha256
+            or locked.locked_guide_sha256 != evidence.locked_guide_sha256
+            or locked.effective_policy_id != UUID(evidence.effective_policy_id)
+            or locked.effective_policy_sha256 != evidence.locked_artifact_policy_sha256
+            or locked.pre_submit_policy_id != UUID(evidence.pre_submit_policy_id)
+            or locked.pre_submit_policy_sha256 != evidence.locked_checker_policy_sha256
+        ):
+            raise ArtifactAdmissionRelationshipError("submission bundle locked context changed")
+        operation_identity = canonical_json_hash(
+            {
+                "request_type": "submission_bundle",
+                "pre_submit_evidence_set_id": evidence.id,
+            }
+        )
+        authority_facts = SubmissionBundleDurableIntentAuthorityFacts(
+            actor_profile_id=UUID(evidence.actor_profile_id),
+            identity_link_id=UUID(evidence.identity_link_id),
+            project_id=UUID(evidence.project_id),
+            task_id=UUID(evidence.task_id),
+            assignment_id=UUID(evidence.assignment_id),
+            predecessor_submission_id=(
+                UUID(evidence.predecessor_submission_id)
+                if evidence.predecessor_submission_id is not None
+                else None
+            ),
+            predecessor_submission_version=evidence.predecessor_submission_version,
+            pre_submit_evidence_set_id=UUID(evidence.id),
+            prepared_generation_id=UUID(evidence.prepared_generation_id),
+            guide_id=UUID(evidence.guide_id),
+            guide_version=evidence.guide_version,
+            source_snapshot_id=UUID(evidence.source_snapshot_id),
+            source_snapshot_sha256=evidence.source_snapshot_sha256,
+            effective_policy_id=UUID(evidence.effective_policy_id),
+            effective_policy_sha256=evidence.locked_artifact_policy_sha256,
+            pre_submit_policy_id=UUID(evidence.pre_submit_policy_id),
+            pre_submit_policy_sha256=evidence.locked_checker_policy_sha256,
+            effective_plan_sha256=evidence.effective_plan_sha256,
+            semantic_manifest_id=UUID(evidence.semantic_manifest_id),
+            semantic_manifest_sha256=evidence.semantic_manifest_sha256,
+            archive_sha256=evidence.archive_sha256,
+            archive_byte_count=evidence.archive_byte_count,
+            media_type=commitment.media_type,
+            storage_scheme=evidence.storage_scheme,
+            operation_identity=operation_identity,
+            replay_durable_intent_id=request.replay_durable_intent_id,
+        )
+        return authority_facts, _AdmissionFacts(
+            request_type="submission_bundle",
+            producer_type="actor_profile",
+            producer_ref=evidence.actor_profile_id,
+            project_id=evidence.project_id,
+            guide_id=evidence.guide_id,
+            task_id=evidence.task_id,
+            guide_source_item_id=None,
+            guide_source_snapshot_id=evidence.source_snapshot_id,
+            checker_run_id=None,
+            logical_role=None,
+            pre_submit_evidence_set_id=evidence.id,
             operation_identity=operation_identity,
         )
 
diff --git a/backend/app/modules/artifacts/submission_admission.py b/backend/app/modules/artifacts/submission_admission.py
new file mode 100644
index 00000000..4db1462c
--- /dev/null
+++ b/backend/app/modules/artifacts/submission_admission.py
@@ -0,0 +1,134 @@
+"""Hidden evidence-bound durable put handoff for one checked submission ZIP."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.modules.artifacts.pre_submit_evidence import PreSubmitPassCapability
+from app.modules.artifacts.schemas import (
+    ArtifactAdmissionResult,
+    SubmissionBundleArtifactAdmissionRequest,
+)
+from app.modules.artifacts.submission_authorization import (
+    SubmissionBundlePreparedAuthorization,
+)
+from app.modules.artifacts.service import (
+    ArtifactAdmissionService,
+    ArtifactStorageOrchestrator,
+)
+from app.modules.artifacts.sources import PreparedArtifact
+from app.modules.artifacts.submission_custody import SubmissionBundlePreparedCustody
+from app.modules.authorization.prepared import PreparedAuthorizationHandle
+
+
+@dataclass(frozen=True, slots=True)
+class SubmissionBundleDurablePutRequest:
+    """Exact live custody and opaque authority for one final durable handoff."""
+
+    prepared_authorization: PreparedAuthorizationHandle
+    prepared_artifact: PreparedArtifact
+    pass_capability: PreSubmitPassCapability
+    replay_durable_intent_id: UUID | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class SubmissionBundleDurablePutResult:
+    """Bounded durable operation result without provider coordinates."""
+
+    put_attempt_id: UUID
+    pre_submit_evidence_set_id: UUID
+    operation_identity: str
+    status: str
+    replayed: bool
+
+
+class SubmissionBundleDurablePutService:
+    """Commit final authority and intent before invoking the generic provider path."""
+
+    def __init__(
+        self,
+        *,
+        session: AsyncSession,
+        admission: ArtifactAdmissionService,
+        storage: ArtifactStorageOrchestrator,
+        authorization: SubmissionBundlePreparedAuthorization,
+    ) -> None:
+        self._session = session
+        self._admission = admission
+        self._storage = storage
+        self._authorization = authorization
+
+    async def admit_in_transaction(
+        self,
+        request: SubmissionBundleDurablePutRequest,
+    ) -> tuple[PreparedArtifact, UUID, ArtifactAdmissionResult]:
+        """Consume live custody and persist the complete intent in the caller transaction."""
+        if type(request) is not SubmissionBundleDurablePutRequest:
+            raise TypeError("invalid submission bundle durable put request")
+        transaction = self._session.sync_session.get_transaction()
+        prepared = request.prepared_artifact
+        if (
+            transaction is None
+            or not transaction.is_active
+            or self._session.in_nested_transaction()
+            or type(request.prepared_authorization) is not PreparedAuthorizationHandle
+            or type(prepared) is not PreparedArtifact
+            or type(request.pass_capability) is not PreSubmitPassCapability
+        ):
+            if type(prepared) is PreparedArtifact:
+                await prepared.close()
+            raise RuntimeError("submission bundle durable transaction is unavailable")
+        try:
+            evidence_set_id = request.pass_capability.evidence_set_id
+            custody = SubmissionBundlePreparedCustody._from_live_preparation(
+                prepared=prepared,
+                capability=request.pass_capability,
+            )
+            admission = await self._admission.admit(
+                SubmissionBundleArtifactAdmissionRequest(
+                    pre_submit_evidence_set_id=evidence_set_id,
+                    custody=custody,
+                    replay_durable_intent_id=request.replay_durable_intent_id,
+                ),
+                submission_prepared_authorization=self._authorization,
+                prepared_authorization=request.prepared_authorization,
+                existing_transaction=True,
+            )
+            return prepared, evidence_set_id, admission
+        except BaseException:
+            await prepared.close()
+            raise
+
+    async def publish_after_commit(
+        self,
+        prepared: PreparedArtifact,
+        evidence_set_id: UUID,
+        admission: ArtifactAdmissionResult,
+    ) -> SubmissionBundleDurablePutResult:
+        """Hand the exact ZIP to storage only after the durable transaction ended."""
+        if self._session.in_transaction():
+            await prepared.close()
+            raise RuntimeError("submission bundle durable transaction is still active")
+        try:
+            if admission.replayed:
+                status = await self._storage.resume_committed_put(
+                    attempt_id=admission.attempt_id,
+                    source=prepared.committed_source,
+                )
+            else:
+                status = await self._storage.execute_committed_put(
+                    attempt_id=admission.attempt_id,
+                    source=prepared.committed_source,
+                )
+            return SubmissionBundleDurablePutResult(
+                put_attempt_id=admission.attempt_id,
+                pre_submit_evidence_set_id=evidence_set_id,
+                operation_identity=admission.operation_identity,
+                status=status,
+                replayed=admission.replayed,
+            )
+        finally:
+            await prepared.close()
diff --git a/backend/app/modules/artifacts/submission_authorization.py b/backend/app/modules/artifacts/submission_authorization.py
new file mode 100644
index 00000000..5c72774d
--- /dev/null
+++ b/backend/app/modules/artifacts/submission_authorization.py
@@ -0,0 +1,35 @@
+"""Internal opaque authorization seam for submission-bundle durable intent."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from app.modules.artifacts.schemas import (
+    ArtifactAuthorityDeniedError,
+    SubmissionBundleDurableIntentAuthorityFacts,
+)
+from app.modules.authorization.prepared import PreparedAuthorizationHandle
+
+
+class SubmissionBundlePreparedAuthorization(Protocol):
+    """Consume final contributor authority in the durable intent transaction."""
+
+    async def consume(
+        self,
+        *,
+        prepared_authorization: PreparedAuthorizationHandle,
+        facts: SubmissionBundleDurableIntentAuthorityFacts,
+    ) -> None: ...
+
+
+class DenySubmissionBundlePreparedAuthorization:
+    """Keep contributor durable preparation unavailable until XINT-05A."""
+
+    async def consume(
+        self,
+        *,
+        prepared_authorization: PreparedAuthorizationHandle,
+        facts: SubmissionBundleDurableIntentAuthorityFacts,
+    ) -> None:
+        del prepared_authorization, facts
+        raise ArtifactAuthorityDeniedError("submission bundle durable preparation is unavailable")
diff --git a/backend/app/modules/artifacts/submission_custody.py b/backend/app/modules/artifacts/submission_custody.py
new file mode 100644
index 00000000..11c58510
--- /dev/null
+++ b/backend/app/modules/artifacts/submission_custody.py
@@ -0,0 +1,52 @@
+"""Sealed process-local custody for one checked submission continuation."""
+
+from __future__ import annotations
+
+from typing import final
+from uuid import UUID
+
+from app.modules.artifacts.pre_submit_evidence import PreSubmitPassCapability
+from app.modules.artifacts.sources import CommittedArtifactSource, PreparedArtifact
+
+
+@final
+class SubmissionBundlePreparedCustody:
+    """Unforgeable join of live prepared bytes and their fresh pass capability."""
+
+    __slots__ = ("_capability", "_prepared")
+
+    def __init__(self, *_: object, **__: object) -> None:
+        raise TypeError("submission bundle custody requires live prepared work")
+
+    @classmethod
+    def _from_live_preparation(
+        cls,
+        *,
+        prepared: PreparedArtifact,
+        capability: PreSubmitPassCapability,
+    ) -> SubmissionBundlePreparedCustody:
+        if (
+            type(prepared) is not PreparedArtifact
+            or type(capability) is not PreSubmitPassCapability
+        ):
+            raise TypeError("submission bundle custody requires live prepared work")
+        capability._assert_live_prepared_custody(
+            prepared_generation_id=prepared.generation_id,
+            archive_sha256=prepared.commitment.sha256,
+        )
+        custody = object.__new__(cls)
+        custody._prepared = prepared
+        custody._capability = capability
+        return custody
+
+    @property
+    def prepared_generation_id(self) -> UUID:
+        return self._prepared.generation_id
+
+    @property
+    def pass_capability(self) -> PreSubmitPassCapability:
+        return self._capability
+
+    @property
+    def source(self) -> CommittedArtifactSource:
+        return self._prepared.committed_source
diff --git a/backend/app/modules/tasks/pre_submit_context.py b/backend/app/modules/tasks/pre_submit_context.py
index c93134c8..0145a262 100644
--- a/backend/app/modules/tasks/pre_submit_context.py
+++ b/backend/app/modules/tasks/pre_submit_context.py
@@ -51,34 +51,36 @@ async def load_locked_pre_submit_context(
     task_id: UUID,
     assignment_id: UUID,
     predecessor_submission_id: UUID | None,
+    include_actor_identity_locks: bool = True,
 ) -> LockedPreSubmitContext:
     """Lock and revalidate task, assignment, predecessor, guide, and policy lineage."""
-    actor_profile = await session.scalar(
-        select(ActorProfile)
-        .where(ActorProfile.id == str(actor_profile_id))
-        .with_for_update()
-    )
-    identity_link = await session.scalar(
-        select(ActorIdentityLink)
-        .where(ActorIdentityLink.id == str(identity_link_id))
-        .with_for_update()
-    )
+    actor_profile = identity_link = None
+    if include_actor_identity_locks:
+        actor_profile = await session.scalar(
+            select(ActorProfile).where(ActorProfile.id == str(actor_profile_id)).with_for_update()
+        )
+        identity_link = await session.scalar(
+            select(ActorIdentityLink)
+            .where(ActorIdentityLink.id == str(identity_link_id))
+            .with_for_update()
+        )
     task = await session.scalar(
-        select(WorkstreamTask)
-        .where(WorkstreamTask.id == str(task_id))
-        .with_for_update()
+        select(WorkstreamTask).where(WorkstreamTask.id == str(task_id)).with_for_update()
     )
     assignment = await session.scalar(
-        select(TaskAssignment)
-        .where(TaskAssignment.id == str(assignment_id))
-        .with_for_update()
+        select(TaskAssignment).where(TaskAssignment.id == str(assignment_id)).with_for_update()
     )
     if (
-        actor_profile is None
-        or actor_profile.status != "active"
-        or identity_link is None
-        or identity_link.actor_profile_id != str(actor_profile_id)
-        or identity_link.status != "active"
+        (
+            include_actor_identity_locks
+            and (
+                actor_profile is None
+                or actor_profile.status != "active"
+                or identity_link is None
+                or identity_link.actor_profile_id != str(actor_profile_id)
+                or identity_link.status != "active"
+            )
+        )
         or task is None
         or assignment is None
         or assignment.task_id != str(task_id)
@@ -121,8 +123,7 @@ async def load_locked_pre_submit_context(
             EffectiveProjectSubmissionArtifactPolicy.id
             == task.locked_effective_project_submission_artifact_policy_id,
             EffectiveProjectSubmissionArtifactPolicy.project_id == task.project_id,
-            EffectiveProjectSubmissionArtifactPolicy.guide_version
-            == task.locked_guide_version,
+            EffectiveProjectSubmissionArtifactPolicy.guide_version == task.locked_guide_version,
             EffectiveProjectSubmissionArtifactPolicy.effective_policy_hash
             == task.locked_effective_project_submission_artifact_policy_hash,
         )
diff --git a/backend/scripts/run_test_lanes.py b/backend/scripts/run_test_lanes.py
index 2b95833c..fbc3940c 100644
--- a/backend/scripts/run_test_lanes.py
+++ b/backend/scripts/run_test_lanes.py
@@ -133,6 +133,7 @@ class TestLane:
             "tests/test_auth.py",
             "tests/test_authorization.py",
             "tests/test_artifact_admission.py",
+            "tests/test_submission_bundle_admission.py",
             "tests/test_artifact_operator_api.py",
             "tests/test_artifact_recovery.py",
             "tests/test_db_session.py",
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index 5f7d4519..a996952a 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -21,7 +21,7 @@
 from scripts.run_isolated_tests import LOOPBACK, NAME_RE, ROLE_RE
 
 DDL_LOCK_DIRECTORY = Path("/tmp")
-EXPECTED_PUBLIC_SCHEMA_SHA256 = "1d427ad3949f452b22c978dc757b6029bb98e524ab4459791cd2a913f4fa5e11"
+EXPECTED_PUBLIC_SCHEMA_SHA256 = "5f2f2345f26361ba82bb8da2a8ee098223b4726d39070df58b704387284c3a1e"
 PROTECTED_TEST_TABLES = (
     "actor_profile_migration_state",
     "alembic_version",
@@ -94,6 +94,7 @@
     "review_queue_entries",
     "submission_policy_mutation_idempotency_records",
     "submission_artifact_policies",
+    "submission_bundle_durable_intents",
     "submissions",
     "task_assignments",
     "workstream_tasks",
@@ -123,6 +124,7 @@
     "review_queue_entries",
     "review_policies",
     "revision_policies",
+    "submission_bundle_durable_intents",
     "submission_policy_mutation_idempotency_records",
 )
 TestDatabaseReset = Callable[..., Awaitable[None]]
diff --git a/backend/tests/test_alembic.py b/backend/tests/test_alembic.py
index 85980e64..23346b68 100644
--- a/backend/tests/test_alembic.py
+++ b/backend/tests/test_alembic.py
@@ -73,7 +73,7 @@
     snapshot_existing_service_rows,
 )
 
-HEAD_REVISION = "0059_policy_execution_claim"
+HEAD_REVISION = "0060_submission_bundle_intent"
 
 pytestmark = pytest.mark.postgres_schema_contract
 
@@ -196,6 +196,102 @@ def test_0058_pre_submit_evidence_empty_round_trip(
     }
 
 
+async def _submission_bundle_intent_schema(database_url: str) -> dict[str, object]:
+    engine = create_async_engine(database_url)
+    try:
+        async with engine.connect() as connection:
+            table_exists = bool(
+                await connection.scalar(
+                    text("select to_regclass('submission_bundle_durable_intents') is not null")
+                )
+            )
+            request_type = await connection.scalar(
+                text(
+                    "select pg_get_constraintdef(oid) from pg_constraint "
+                    "where conrelid='artifact_put_attempts'::regclass "
+                    "and conname='ck_artifact_put_attempts_producer_request_type'"
+                )
+            )
+            constraints = {
+                row
+                for row in (
+                    await connection.execute(
+                        text(
+                            "select conname from pg_constraint where conrelid="
+                            "to_regclass('submission_bundle_durable_intents')"
+                        )
+                    )
+                ).scalars()
+            }
+            triggers = {
+                row
+                for row in (
+                    await connection.execute(
+                        text(
+                            "select tgname from pg_trigger where not tgisinternal and "
+                            "tgrelid=to_regclass('submission_bundle_durable_intents')"
+                        )
+                    )
+                ).scalars()
+            }
+            receipt_triggers = {
+                row
+                for row in (
+                    await connection.execute(
+                        text(
+                            "select tgname from pg_trigger where not tgisinternal and "
+                            "tgrelid='artifact_operation_receipts'::regclass"
+                        )
+                    )
+                ).scalars()
+            }
+            return {
+                "table_exists": table_exists,
+                "request_type": request_type,
+                "constraints": constraints,
+                "triggers": triggers,
+                "receipt_triggers": receipt_triggers,
+            }
+    finally:
+        await engine.dispose()
+
+
+def test_0060_submission_bundle_intent_empty_round_trip(
+    isolated_database_env: str,
+    migration_lock,
+) -> None:
+    config = _alembic_config()
+    with migration_lock():
+        try:
+            command.downgrade(config, "0059_policy_execution_claim")
+            prior = asyncio.run(_submission_bundle_intent_schema(isolated_database_env))
+            command.upgrade(config, HEAD_REVISION)
+            installed = asyncio.run(_submission_bundle_intent_schema(isolated_database_env))
+            command.downgrade(config, "0059_policy_execution_claim")
+            restored = asyncio.run(_submission_bundle_intent_schema(isolated_database_env))
+            command.upgrade(config, HEAD_REVISION)
+            repeated = asyncio.run(_submission_bundle_intent_schema(isolated_database_env))
+        finally:
+            command.upgrade(config, "head")
+
+    assert prior == restored
+    assert prior["table_exists"] is False
+    assert "submission_bundle" not in str(prior["request_type"])
+    assert installed == repeated
+    assert installed["table_exists"] is True
+    assert "submission_bundle" in str(installed["request_type"])
+    assert {
+        "uq_submission_bundle_intent_evidence",
+        "uq_submission_bundle_intent_put_attempt",
+    }.issubset(installed["constraints"])
+    assert installed["triggers"] == {
+        "submission_bundle_durable_intent_put_attempt",
+        "submission_bundle_durable_intents_immutable",
+        "submission_bundle_durable_intents_no_truncate",
+    }
+    assert "artifact_receipt_producer_reference" in installed["receipt_triggers"]
+
+
 async def _submission_policy_authority_shape(database_url: str) -> dict[str, object]:
     engine = create_async_engine(database_url)
     try:
diff --git a/backend/tests/test_artifact_architecture.py b/backend/tests/test_artifact_architecture.py
index 320ba897..abdca7b4 100644
--- a/backend/tests/test_artifact_architecture.py
+++ b/backend/tests/test_artifact_architecture.py
@@ -72,6 +72,7 @@
     "ArtifactAdmissionResult",
     "CheckerOutputArtifactAdmissionRequest",
     "GuideArtifactAdmissionRequest",
+    "SubmissionBundleArtifactAdmissionRequest",
 }
 
 RETIRED_CONTRIBUTOR_INTAKE_NAMES = {
diff --git a/backend/tests/test_artifact_authorization.py b/backend/tests/test_artifact_authorization.py
index 291ce106..c2ce1d76 100644
--- a/backend/tests/test_artifact_authorization.py
+++ b/backend/tests/test_artifact_authorization.py
@@ -191,6 +191,7 @@ async def test_quota_reconciliation_is_configuration_driven_and_rollback_safe()
         guide_source_snapshot_id=None,
         checker_run_id=None,
         logical_role=None,
+        pre_submit_evidence_set_id=None,
         operation_identity="sha256:" + "a" * 64,
     )
     result = await service._reserve_charges(
diff --git a/backend/tests/test_ci_test_lanes.py b/backend/tests/test_ci_test_lanes.py
index 7909b23d..33dd821c 100644
--- a/backend/tests/test_ci_test_lanes.py
+++ b/backend/tests/test_ci_test_lanes.py
@@ -59,6 +59,7 @@ def test_measured_hotspots_have_explicit_semantic_owners() -> None:
     assert {
         "tests/test_actors.py",
         "tests/test_artifact_admission.py",
+        "tests/test_submission_bundle_admission.py",
         "tests/test_authorization.py",
         "tests/test_guide_artifacts.py",
         "tests/test_mutation_policy.py",
diff --git a/backend/tests/test_default_pre_submit_execution.py b/backend/tests/test_default_pre_submit_execution.py
index 36808cd8..2531dbd5 100644
--- a/backend/tests/test_default_pre_submit_execution.py
+++ b/backend/tests/test_default_pre_submit_execution.py
@@ -9,15 +9,17 @@
 import threading
 from types import SimpleNamespace
 from typing import Any, cast
+from unittest.mock import AsyncMock
 import zipfile
-from uuid import uuid4
+from uuid import UUID, uuid4
 
 import pytest
-from sqlalchemy import text
+from sqlalchemy import func, select, text
 from sqlalchemy.exc import DBAPIError
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 
 from app.core.hashing import canonical_json_hash
+from app.core.config import Settings
 from app.interfaces.artifact_operations import PreparedBundleMaterializationRequest
 from app.modules.artifacts.preparation import (
     HARD_MAXIMUM_ARTIFACT_BYTES,
@@ -27,6 +29,19 @@
     ArtifactScratchManager,
 )
 from app.modules.artifacts.schemas import ArtifactAuthorityDeniedError
+from app.modules.artifacts.models import SubmissionBundleDurableIntent
+from app.modules.artifacts.service import (
+    ArtifactAdmissionRelationshipError,
+    ArtifactAdmissionService,
+    ArtifactStorageNamespaceSpec,
+)
+from app.modules.artifacts.submission_admission import (
+    SubmissionBundleDurablePutRequest,
+    SubmissionBundleDurablePutService,
+)
+from app.modules.artifacts.submission_authorization import (
+    DenySubmissionBundlePreparedAuthorization,
+)
 from app.modules.artifacts.submission_archive import (
     SubmissionArchiveInspector,
     SubmissionArchiveLimits,
@@ -56,12 +71,24 @@
     SubmissionPacketView,
     validate_pre_submission_execution_result,
 )
+from tests.artifact_store_helpers import artifact_admission_limit_settings
 
 
 async def _bytes(value: bytes):
     yield value
 
 
+class _AllowSubmissionPreparedAuthorization:
+    """Test-only final authority that records transaction-bound consumption."""
+
+    def __init__(self) -> None:
+        self.facts = None
+
+    async def consume(self, *, prepared_authorization, facts) -> None:
+        assert type(prepared_authorization) is PreparedAuthorizationHandle
+        self.facts = facts
+
+
 @pytest.mark.asyncio
 async def test_evidence_workflow_requires_transaction_free_session() -> None:
     materialization = SimpleNamespace(materialize_prepared_bundle=lambda _request: None)
@@ -82,11 +109,16 @@ async def test_evidence_workflow_requires_transaction_free_session() -> None:
 def _archive(path: str = "task.toml", *, extra_path: str | None = None) -> bytes:
     output = BytesIO()
     with zipfile.ZipFile(output, "w") as archive:
-        archive.writestr(path, b"[task]\nname='proof'\n")
+
+        def write(name: str, value: bytes) -> None:
+            entry = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
+            archive.writestr(entry, value)
+
+        write(path, b"[task]\nname='proof'\n")
         if extra_path is not None:
-            archive.writestr(extra_path, b"blocked\n")
+            write(extra_path, b"blocked\n")
         if path != "evidence/results":
-            archive.writestr("evidence/results", b"verified\n")
+            write("evidence/results", b"verified\n")
     return output.getvalue()
 
 
@@ -398,6 +430,9 @@ async def test_effective_evidence_workflow_persists_once_and_replays_exactly(
         ("revision_policies", "revision_policy_mutation_custody"),
     )
     blocked_prepared = None
+    replay_prepared = None
+    drift_prepared = None
+    denied_prepared = None
     original_prepared_closed = False
     tables = (
         "artifact_contents",
@@ -595,6 +630,35 @@ async def test_effective_evidence_workflow_persists_once_and_replays_exactly(
                     storage_scheme="s3",
                 ),
             )
+
+            async def fresh_checked_bundle():
+                prepared = await preparation.prepare(
+                    _bytes(_archive()),
+                    media_type="application/zip",
+                )
+                inspection = await prepared.inspect(inspector)
+                manifest = build_submission_manifest(inspection)
+                fresh_request = replace(
+                    request,
+                    prepared_artifact=prepared,
+                    inspection=inspection,
+                    manifest=manifest,
+                    change_gate=evaluate_submission_change(
+                        commitment=prepared.commitment,
+                        manifest=manifest,
+                        predecessor=None,
+                        predecessor_exists=False,
+                    ),
+                )
+                result = await workflow.execute(
+                    fresh_request,
+                    actor_profile_id=actor_id,
+                    identity_link_id=identity_link_id,
+                    predecessor_submission_id=None,
+                )
+                assert result.pass_capability is not None
+                return prepared, result
+
             first = await workflow.execute(
                 request,
                 actor_profile_id=actor_id,
@@ -607,8 +671,163 @@ async def test_effective_evidence_workflow_persists_once_and_replays_exactly(
                 identity_link_id=identity_link_id,
                 predecessor_submission_id=None,
             )
-            await request.prepared_artifact.close()
+            assert first.pass_capability is not None
+            namespace = ArtifactStorageNamespaceSpec(
+                backend="local",
+                adapter="local",
+                provider_profile="test",
+                namespace_descriptor={"test": "submission-bundle"},
+                namespace_fingerprint=canonical_json_hash({"test": "submission-bundle"}),
+            )
+            admission_settings = Settings(
+                **artifact_admission_limit_settings(1024 * 1024),
+                environment="test",
+                artifact_store_backend="local",
+                artifact_local_root=tmp_path / "durable",
+                artifact_scratch_root=tmp_path / "scratch",
+                artifact_scratch_minimum_free_bytes=0,
+            )
+            provider = SimpleNamespace(
+                execute_committed_put=AsyncMock(),
+                resume_committed_put=AsyncMock(),
+            )
+            final_authority = _AllowSubmissionPreparedAuthorization()
+            durable_service = SubmissionBundleDurablePutService(
+                session=session,
+                admission=ArtifactAdmissionService(
+                    session,
+                    admission_settings,
+                    namespace,
+                ),
+                storage=provider,
+                authorization=final_authority,
+            )
+            async with session.begin():
+                (
+                    retained,
+                    selected_evidence_id,
+                    first_admission,
+                ) = await durable_service.admit_in_transaction(
+                    SubmissionBundleDurablePutRequest(
+                        prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                        prepared_artifact=request.prepared_artifact,
+                        pass_capability=first.pass_capability,
+                    )
+                )
+            provider.execute_committed_put.assert_not_awaited()
+            provider.resume_committed_put.assert_not_awaited()
+            intent = await session.scalar(
+                select(SubmissionBundleDurableIntent).where(
+                    SubmissionBundleDurableIntent.put_attempt_id == str(first_admission.attempt_id)
+                )
+            )
+            assert intent is not None
+            replay_intent_id = UUID(intent.id)
+            await session.rollback()
+            assert retained is request.prepared_artifact
+            await retained.close()
             original_prepared_closed = True
+
+            replay_prepared, fresh = await fresh_checked_bundle()
+            async with session.begin():
+                (
+                    replay_retained,
+                    replay_evidence_id,
+                    replay_admission,
+                ) = await durable_service.admit_in_transaction(
+                    SubmissionBundleDurablePutRequest(
+                        prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                        prepared_artifact=replay_prepared,
+                        pass_capability=fresh.pass_capability,
+                        replay_durable_intent_id=replay_intent_id,
+                    )
+                )
+            assert replay_admission.replayed is True
+            assert replay_admission.attempt_id == first_admission.attempt_id
+            assert replay_evidence_id == fresh.evidence.evidence_set_id
+            await replay_retained.close()
+            intent_count = int(
+                await session.scalar(
+                    select(func.count()).select_from(SubmissionBundleDurableIntent)
+                )
+                or 0
+            )
+            assert intent_count == 1
+            await session.rollback()
+
+            drift_prepared, drift = await fresh_checked_bundle()
+            await session.execute(
+                text("update task_assignments set status='inactive' where id=:assignment"),
+                {"assignment": str(request.assignment_id)},
+            )
+            await session.commit()
+            with pytest.raises(
+                ArtifactAdmissionRelationshipError,
+                match="passing evidence is unavailable|locked_context",
+            ):
+                async with session.begin():
+                    await durable_service.admit_in_transaction(
+                        SubmissionBundleDurablePutRequest(
+                            prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                            prepared_artifact=drift_prepared,
+                            pass_capability=drift.pass_capability,
+                        )
+                    )
+            await session.execute(
+                text("update task_assignments set status='active' where id=:assignment"),
+                {"assignment": str(request.assignment_id)},
+            )
+            await session.commit()
+
+            denied_prepared, denied = await fresh_checked_bundle()
+            denied_service = SubmissionBundleDurablePutService(
+                session=session,
+                admission=ArtifactAdmissionService(
+                    session,
+                    admission_settings,
+                    namespace,
+                ),
+                storage=provider,
+                authorization=DenySubmissionBundlePreparedAuthorization(),
+            )
+            with pytest.raises(ArtifactAuthorityDeniedError):
+                async with session.begin():
+                    await denied_service.admit_in_transaction(
+                        SubmissionBundleDurablePutRequest(
+                            prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                            prepared_artifact=denied_prepared,
+                            pass_capability=denied.pass_capability,
+                        )
+                    )
+            durable_counts = {
+                table: int(await session.scalar(text(f"select count(*) from {table}")) or 0)
+                for table in (
+                    "artifact_put_attempts",
+                    "submission_bundle_durable_intents",
+                    "artifact_admission_charges",
+                )
+            }
+            guide_continuation_matches = int(
+                await session.scalar(
+                    text(
+                        "select count(*) from artifact_put_attempts attempt "
+                        "join guide_source_snapshot_items item "
+                        "on item.id=attempt.guide_source_item_id "
+                        "where attempt.producer_request_type='submission_bundle'"
+                    )
+                )
+                or 0
+            )
+            await session.rollback()
+            assert durable_counts == {
+                "artifact_put_attempts": 1,
+                "submission_bundle_durable_intents": 1,
+                "artifact_admission_charges": 4,
+            }
+            assert guide_continuation_matches == 0
+            provider.execute_committed_put.assert_not_awaited()
+            provider.resume_committed_put.assert_not_awaited()
+            assert selected_evidence_id == first.evidence.evidence_set_id
             blocked_prepared = await preparation.prepare(
                 _bytes(_archive("task.toml")), media_type="application/zip"
             )
@@ -701,15 +920,27 @@ async def test_effective_evidence_workflow_persists_once_and_replays_exactly(
                         )
                     )
     finally:
-        if blocked_prepared is not None:
-            await blocked_prepared.close()
+        for prepared in (
+            blocked_prepared,
+            replay_prepared,
+            drift_prepared,
+            denied_prepared,
+        ):
+            if prepared is not None:
+                await prepared.close()
         if not original_prepared_closed:
             await request.prepared_artifact.close()
-        manager.close()
-        async with engine.begin() as connection:
-            for table, trigger in reversed(custody_triggers):
-                await connection.execute(text(f"alter table {table} enable trigger {trigger}"))
-        await engine.dispose()
+        try:
+            manager.close()
+        finally:
+            try:
+                async with engine.begin() as connection:
+                    for table, trigger in reversed(custody_triggers):
+                        await connection.execute(
+                            text(f"alter table {table} enable trigger {trigger}")
+                        )
+            finally:
+                await engine.dispose()
 
     assert first.evidence.replayed is False
     assert replay.evidence.replayed is True
@@ -723,9 +954,12 @@ async def test_effective_evidence_workflow_persists_once_and_replays_exactly(
     assert blocked.failure_audit["event_type"] == "pre_submission_check_failed"
     assert blocked.failure_audit["failed_count"] >= 1
     assert "task.toml" not in repr(blocked.failure_audit)
-    assert evidence_count == 2
-    assert result_count == 2 * len(request.effective_plan.entries)
-    assert after == before
+    assert evidence_count == 5
+    assert result_count == 5 * len(request.effective_plan.entries)
+    assert after == {
+        **before,
+        "artifact_put_attempts": before["artifact_put_attempts"] + 1,
+    }
 
 
 @pytest.mark.asyncio
diff --git a/backend/tests/test_effective_pre_submit_execution.py b/backend/tests/test_effective_pre_submit_execution.py
index 608bd4b2..6ff13c33 100644
--- a/backend/tests/test_effective_pre_submit_execution.py
+++ b/backend/tests/test_effective_pre_submit_execution.py
@@ -15,7 +15,7 @@
 from app.modules.artifacts.pre_submit_evidence import (
     PreSubmitEvidenceConflict,
     PreSubmitEvidenceContext,
-    PreSubmitPassCapability,
+    PreSubmitEvidenceService,
     PersistedPreSubmitEvidence,
     pre_submit_failure_audit_payload,
     semantic_manifest_identity,
@@ -89,7 +89,7 @@ def test_semantic_manifest_identity_is_server_deterministic() -> None:
 def test_pass_capability_is_generation_bound_and_single_use() -> None:
     evidence_set_id = uuid4()
     generation_id = uuid4()
-    capability = PreSubmitPassCapability(
+    capability = PreSubmitEvidenceService(SimpleNamespace())._mint_pass_capability(
         evidence_set_id=evidence_set_id,
         prepared_generation_id=generation_id,
         predecessor_submission_id=None,
diff --git a/backend/tests/test_submission_bundle_admission.py b/backend/tests/test_submission_bundle_admission.py
new file mode 100644
index 00000000..3c4fa4bf
--- /dev/null
+++ b/backend/tests/test_submission_bundle_admission.py
@@ -0,0 +1,353 @@
+"""Focused custody and ordering proof for submission-bundle durable intent."""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+from uuid import uuid4
+
+import pytest
+
+from app.modules.artifacts.pre_submit_evidence import (
+    PreSubmitEvidenceConflict,
+    PreSubmitEvidenceService,
+    PreSubmitPassCapability,
+)
+from app.modules.artifacts.preparation import ArtifactPreparationService, ArtifactScratchManager
+from app.modules.artifacts.schemas import (
+    ArtifactAdmissionResult,
+    ArtifactAuthorityDeniedError,
+    SubmissionBundleArtifactAdmissionRequest,
+)
+from app.modules.artifacts.submission_authorization import (
+    DenySubmissionBundlePreparedAuthorization,
+)
+from app.modules.artifacts.submission_admission import (
+    SubmissionBundleDurablePutRequest,
+    SubmissionBundleDurablePutService,
+)
+from app.modules.artifacts.service import ArtifactAdmissionService
+from app.modules.artifacts.submission_custody import SubmissionBundlePreparedCustody
+from app.modules.authorization.prepared import PreparedAuthorizationHandle
+from tests.artifact_store_helpers import artifact_byte_stream, artifact_preparation_limits
+
+
+def _sha(character: str) -> str:
+    return "sha256:" + character * 64
+
+
+async def _prepared(tmp_path):
+    manager = ArtifactScratchManager(
+        root=tmp_path / "scratch",
+        limits=artifact_preparation_limits(),
+    )
+    prepared = await ArtifactPreparationService(manager).prepare(
+        artifact_byte_stream(b"PK\x03\x04checked submission"),
+        media_type="application/zip",
+    )
+    return manager, prepared
+
+
+def _capability(prepared, evidence_set_id):
+    service = PreSubmitEvidenceService(SimpleNamespace())
+    return service._mint_pass_capability(
+        evidence_set_id=evidence_set_id,
+        prepared_generation_id=prepared.generation_id,
+        predecessor_submission_id=None,
+        effective_plan_sha256=_sha("7"),
+        archive_sha256=prepared.commitment.sha256,
+        semantic_manifest_sha256=_sha("2"),
+        storage_scheme="s3",
+    )
+
+
+@pytest.mark.asyncio
+async def test_durable_put_admits_in_transaction_then_publishes(tmp_path) -> None:
+    manager, prepared = await _prepared(tmp_path)
+    evidence_set_id = uuid4()
+    state = {"transaction": True}
+    transaction = SimpleNamespace(is_active=True)
+    session = SimpleNamespace(
+        sync_session=SimpleNamespace(get_transaction=lambda: transaction),
+        in_nested_transaction=lambda: False,
+        in_transaction=lambda: state["transaction"],
+    )
+    admission_result = ArtifactAdmissionResult(
+        attempt_id=uuid4(),
+        status="prepared",
+        operation_identity=_sha("3"),
+        request_digest=_sha("4"),
+        charge_ids=(uuid4(),),
+        replayed=False,
+    )
+
+    async def admit(request, **values):
+        assert state["transaction"] is True
+        assert type(request) is SubmissionBundleArtifactAdmissionRequest
+        assert request.pre_submit_evidence_set_id == evidence_set_id
+        assert request.custody.prepared_generation_id == prepared.generation_id
+        assert type(request.custody.pass_capability) is PreSubmitPassCapability
+        assert values["existing_transaction"] is True
+        return admission_result
+
+    admission = SimpleNamespace(admit=admit)
+    storage = SimpleNamespace(
+        execute_committed_put=AsyncMock(return_value="object_confirmed"),
+        resume_committed_put=AsyncMock(),
+    )
+    service = SubmissionBundleDurablePutService(
+        session=session,
+        admission=admission,
+        storage=storage,
+        authorization=object(),
+    )
+    handle = object.__new__(PreparedAuthorizationHandle)
+    try:
+        retained, selected_evidence_id, durable = await service.admit_in_transaction(
+            SubmissionBundleDurablePutRequest(
+                prepared_authorization=handle,
+                prepared_artifact=prepared,
+                pass_capability=_capability(prepared, evidence_set_id),
+            )
+        )
+        storage.execute_committed_put.assert_not_awaited()
+        state["transaction"] = False
+        result = await service.publish_after_commit(
+            retained,
+            selected_evidence_id,
+            durable,
+        )
+        assert result.pre_submit_evidence_set_id == evidence_set_id
+        assert result.status == "object_confirmed"
+        storage.execute_committed_put.assert_awaited_once()
+        storage.resume_committed_put.assert_not_awaited()
+        with pytest.raises(RuntimeError, match="prepared artifact is closed"):
+            _ = prepared.generation_id
+    finally:
+        await prepared.close()
+        manager.close()
+
+
+def test_pass_capability_rejects_direct_construction() -> None:
+    with pytest.raises(TypeError, match="can only be created by pre-submit evidence"):
+        PreSubmitPassCapability(
+            evidence_set_id=uuid4(),
+            prepared_generation_id=uuid4(),
+            predecessor_submission_id=None,
+            effective_plan_sha256=_sha("7"),
+            archive_sha256=_sha("1"),
+            semantic_manifest_sha256=_sha("2"),
+            storage_scheme="s3",
+        )
+
+
+def test_pass_capability_rejects_unregistered_service_owner() -> None:
+    with pytest.raises(TypeError, match="can only be created by pre-submit evidence"):
+        PreSubmitPassCapability._from_evidence_service(
+            owner=object.__new__(PreSubmitEvidenceService),
+            binding=object(),
+            evidence_set_id=uuid4(),
+            prepared_generation_id=uuid4(),
+            predecessor_submission_id=None,
+            effective_plan_sha256=_sha("7"),
+            archive_sha256=_sha("1"),
+            semantic_manifest_sha256=_sha("2"),
+            storage_scheme="s3",
+        )
+
+
+def test_submission_custody_rejects_direct_construction() -> None:
+    with pytest.raises(TypeError, match="requires live prepared work"):
+        SubmissionBundlePreparedCustody()
+
+
+def test_submission_request_rejects_evidence_without_live_custody() -> None:
+    request = SubmissionBundleArtifactAdmissionRequest(
+        pre_submit_evidence_set_id=uuid4(),
+        custody=object(),
+        replay_durable_intent_id=None,
+    )
+    with pytest.raises(TypeError, match="prepared custody is unavailable"):
+        _ = request.source
+
+
+def test_replay_lineage_includes_policy_catalogue_and_manifest_identity() -> None:
+    values = {
+        "actor_profile_id": str(uuid4()),
+        "identity_link_id": str(uuid4()),
+        "project_id": str(uuid4()),
+        "task_id": str(uuid4()),
+        "assignment_id": str(uuid4()),
+        "predecessor_submission_id": None,
+        "predecessor_submission_version": None,
+        "archive_sha256": _sha("1"),
+        "archive_byte_count": 10,
+        "semantic_manifest_id": str(uuid4()),
+        "semantic_manifest_sha256": _sha("2"),
+        "guide_id": str(uuid4()),
+        "guide_version": "1",
+        "source_snapshot_id": str(uuid4()),
+        "source_snapshot_sha256": _sha("3"),
+        "locked_guide_sha256": _sha("4"),
+        "effective_policy_id": str(uuid4()),
+        "locked_artifact_policy_sha256": _sha("5"),
+        "pre_submit_policy_id": str(uuid4()),
+        "locked_checker_policy_sha256": _sha("6"),
+        "effective_plan_sha256": _sha("7"),
+        "catalogue_id": "workstream.default",
+        "catalogue_version": "1",
+        "catalogue_manifest_sha256": _sha("8"),
+        "storage_scheme": "s3",
+        "terminal_status": "passed",
+        "eligible": True,
+        "result_count": 2,
+        "result_manifest_sha256": _sha("9"),
+    }
+    original = SimpleNamespace(**values)
+    for field, changed in {
+        "semantic_manifest_id": str(uuid4()),
+        "locked_guide_sha256": _sha("a"),
+        "effective_policy_id": str(uuid4()),
+        "pre_submit_policy_id": str(uuid4()),
+        "catalogue_id": "other.catalogue",
+        "catalogue_version": "2",
+        "catalogue_manifest_sha256": _sha("b"),
+    }.items():
+        drifted = SimpleNamespace(**{**values, field: changed})
+        assert ArtifactAdmissionService._submission_replay_lineage(
+            original
+        ) != ArtifactAdmissionService._submission_replay_lineage(drifted)
+
+
+@pytest.mark.asyncio
+async def test_concurrent_pass_capability_consumption_has_one_winner(tmp_path) -> None:
+    manager, prepared = await _prepared(tmp_path)
+    capability = _capability(prepared, uuid4())
+
+    def consume() -> bool:
+        try:
+            capability.consume(
+                prepared_generation_id=prepared.generation_id,
+                predecessor_submission_id=None,
+                effective_plan_sha256=capability.effective_plan_sha256,
+                archive_sha256=prepared.commitment.sha256,
+                semantic_manifest_sha256=capability.semantic_manifest_sha256,
+                storage_scheme=capability.storage_scheme,
+            )
+        except PreSubmitEvidenceConflict:
+            return False
+        return True
+
+    try:
+        outcomes = await asyncio.gather(
+            asyncio.to_thread(consume),
+            asyncio.to_thread(consume),
+        )
+        assert sorted(outcomes) == [False, True]
+    finally:
+        await prepared.close()
+        manager.close()
+
+
+@pytest.mark.asyncio
+async def test_fresh_checked_custody_resumes_existing_committed_intent(tmp_path) -> None:
+    manager, prepared = await _prepared(tmp_path)
+    evidence_set_id = uuid4()
+    replay_intent_id = uuid4()
+    state = {"transaction": True}
+    admission_result = ArtifactAdmissionResult(
+        attempt_id=uuid4(),
+        status="absent_replay_required",
+        operation_identity=_sha("3"),
+        request_digest=_sha("4"),
+        charge_ids=(uuid4(),),
+        replayed=True,
+    )
+
+    async def admit(request, **values):
+        assert request.replay_durable_intent_id == replay_intent_id
+        return admission_result
+
+    storage = SimpleNamespace(
+        execute_committed_put=AsyncMock(),
+        resume_committed_put=AsyncMock(return_value="object_confirmed"),
+    )
+    service = SubmissionBundleDurablePutService(
+        session=SimpleNamespace(
+            sync_session=SimpleNamespace(get_transaction=lambda: SimpleNamespace(is_active=True)),
+            in_nested_transaction=lambda: False,
+            in_transaction=lambda: state["transaction"],
+        ),
+        admission=SimpleNamespace(admit=admit),
+        storage=storage,
+        authorization=object(),
+    )
+    try:
+        retained, selected_evidence_id, durable = await service.admit_in_transaction(
+            SubmissionBundleDurablePutRequest(
+                prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                prepared_artifact=prepared,
+                pass_capability=_capability(prepared, evidence_set_id),
+                replay_durable_intent_id=replay_intent_id,
+            )
+        )
+        state["transaction"] = False
+        result = await service.publish_after_commit(
+            retained,
+            selected_evidence_id,
+            durable,
+        )
+        assert result.replayed is True
+        storage.resume_committed_put.assert_awaited_once()
+        storage.execute_committed_put.assert_not_awaited()
+    finally:
+        await prepared.close()
+        manager.close()
+
+
+@pytest.mark.asyncio
+async def test_durable_put_rejects_capability_replay_before_admission(tmp_path) -> None:
+    manager, prepared = await _prepared(tmp_path)
+    capability = _capability(prepared, uuid4())
+    capability.consume(
+        prepared_generation_id=prepared.generation_id,
+        predecessor_submission_id=None,
+        effective_plan_sha256=capability.effective_plan_sha256,
+        archive_sha256=prepared.commitment.sha256,
+        semantic_manifest_sha256=capability.semantic_manifest_sha256,
+        storage_scheme=capability.storage_scheme,
+    )
+    admission = SimpleNamespace(admit=AsyncMock())
+    service = SubmissionBundleDurablePutService(
+        session=SimpleNamespace(
+            sync_session=SimpleNamespace(get_transaction=lambda: SimpleNamespace(is_active=True)),
+            in_nested_transaction=lambda: False,
+            in_transaction=lambda: True,
+        ),
+        admission=admission,
+        storage=SimpleNamespace(),
+        authorization=object(),
+    )
+    try:
+        with pytest.raises(PreSubmitEvidenceConflict):
+            await service.admit_in_transaction(
+                SubmissionBundleDurablePutRequest(
+                    prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+                    prepared_artifact=prepared,
+                    pass_capability=capability,
+                )
+            )
+        admission.admit.assert_not_awaited()
+    finally:
+        await prepared.close()
+        manager.close()
+
+
+@pytest.mark.asyncio
+async def test_deny_submission_authority_fails_closed() -> None:
+    with pytest.raises(ArtifactAuthorityDeniedError):
+        await DenySubmissionBundlePreparedAuthorization().consume(
+            prepared_authorization=object.__new__(PreparedAuthorizationHandle),
+            facts=object(),
+        )
diff --git a/docs/architecture_data_model.md b/docs/architecture_data_model.md
index aa3867b8..9b6e9d61 100644
--- a/docs/architecture_data_model.md
+++ b/docs/architecture_data_model.md
@@ -814,6 +814,16 @@ execution produces a process-local, generation- and predecessor-bound
 single-use capability for immediate admission continuation; a later attempt
 must re-prepare the bundle. The evidence-set ID alone is never that capability.
 
+ART-04C1 consumes that live capability and exact prepared generation inside the
+final authorization transaction. One immutable `SubmissionBundleDurableIntent`
+then joins the passing evidence set one-to-one with the generic
+`ArtifactPutAttempt`. The join is the database-recoverable producer fence for
+later verification publication; it does not duplicate evidence lineage, store
+scratch state, or create a bindable admission. Provisional capacity, the put
+attempt, authorization evidence, and this join commit before provider I/O.
+Generic observation and recovery continue from the put attempt after process
+loss.
+
 Blocking pre-submit failures prevent submission creation, create no submission
 row, no submission version, no task transition to `submitted`, and no
 submission-created audit event. Workstream still writes a task audit event named
diff --git a/docs/spec_artifact_storage_service.md b/docs/spec_artifact_storage_service.md
index 6992df1f..c094885c 100644
--- a/docs/spec_artifact_storage_service.md
+++ b/docs/spec_artifact_storage_service.md
@@ -1239,8 +1239,8 @@ messages.
 
 Before provider I/O, the final preparation transaction persists one immutable
 submission-producer intent linked by foreign keys to the exact passing
-`PreSubmitEvidenceSet` and generic `ArtifactPutAttempt`. It preserves the actor
-and identity link, project, task, assignment, predecessor identity/version,
+`PreSubmitEvidenceSet` and generic `ArtifactPutAttempt`. Those typed joins
+preserve the actor and identity link, project, task, assignment, predecessor identity/version,
 locked guide/snapshot/policy lineage, semantic-manifest identity/digest,
 archive digest/size/media type, effective-plan digest, storage scheme, and
 operation identity required for database-only verified-admission publication.
@@ -1259,6 +1259,9 @@ execution, which produces a new prepared generation, evidence identity, and
 single-use pass capability. If the process dies after intent commit, the
 existing generic put-attempt observation and recovery machinery owns the
 technical obligation without another submission-specific recovery aggregate.
+Migration `0060_submission_bundle_intent` installs the immutable join
+and extends only the existing generic put-attempt and receipt producer shapes
+needed for submission bundles.
 
 ### SubmissionBundleAdmission