diff --git a/src/proto_alpha/lib_benchmarks_proto/dal_benchmarks.ml b/src/proto_alpha/lib_benchmarks_proto/dal_benchmarks.ml
index 82a1f5aa6a992804af612ed4f14521cc1e3a3b68..8e2f3b535e2f81605fe41e5c79dbb5876338d160 100644
--- a/src/proto_alpha/lib_benchmarks_proto/dal_benchmarks.ml
+++ b/src/proto_alpha/lib_benchmarks_proto/dal_benchmarks.ml
@@ -107,7 +107,12 @@ module Publish_commitment : Benchmark.S = struct
       in
       let workload = () in
       let closure () =
-        match Dal_apply.apply_publish_commitment ctxt op with
+        match
+          Dal_apply.apply_publish_commitment
+            ctxt
+            op
+            ~source:Environment.Signature.Public_key_hash.zero
+        with
         | Error errs ->
             Format.eprintf "%a@." Environment.Error_monad.pp_trace errs ;
             Stdlib.failwith
diff --git a/src/proto_alpha/lib_plugin/RPC.ml b/src/proto_alpha/lib_plugin/RPC.ml
index 56a610eef18eee875adaa1e864503f3dfb1d61fb..9f62c42cf03c334fe8c1a6264340d6761e9537e0 100644
--- a/src/proto_alpha/lib_plugin/RPC.ml
+++ b/src/proto_alpha/lib_plugin/RPC.ml
@@ -3117,7 +3117,7 @@ module Dal = struct
     let level = Option.value level ~default:(Level.current ctxt).level in
     let* result = Dal.Slot.find_slot_headers ctxt level in
     match result with
-    | Some l -> return l
+    | Some l -> return @@ List.map fst l
     | None ->
         Environment.Error_monad.tzfail
         @@ Published_slot_headers_not_initialized level
diff --git a/src/proto_alpha/lib_protocol/alpha_context.mli b/src/proto_alpha/lib_protocol/alpha_context.mli
index eb0311381e2d2763c51e680bb215dc7ef90c9577..d2370934ad697642fcbc85794c1c653d0d0b7a3a 100644
--- a/src/proto_alpha/lib_protocol/alpha_context.mli
+++ b/src/proto_alpha/lib_protocol/alpha_context.mli
@@ -2871,10 +2871,13 @@ module Dal : sig
       val equal : t -> t -> bool
     end
 
-    val register_slot_header : context -> Header.t -> context tzresult
+    val register_slot_header :
+      context -> Header.t -> source:public_key_hash -> context tzresult
 
     val find_slot_headers :
-      context -> Raw_level.t -> Header.t list option tzresult Lwt.t
+      context ->
+      Raw_level.t ->
+      (Header.t * Signature.public_key_hash) list option tzresult Lwt.t
 
     val finalize_current_slot_headers : context -> context Lwt.t
 
@@ -2927,7 +2930,10 @@ module Dal : sig
       t ->
       Raw_level.t ->
       number_of_slots:int ->
-      (Slot.Header.t * Attestation.attestation_status) list ->
+      (Slot.Header.t
+      * Signature.public_key_hash
+      * Attestation.attestation_status)
+      list ->
       t tzresult
 
     val update_skip_list :
@@ -2935,7 +2941,10 @@ module Dal : sig
       History_cache.t ->
       Raw_level.t ->
       number_of_slots:int ->
-      (Slot.Header.t * Attestation.attestation_status) list ->
+      (Slot.Header.t
+      * Signature.public_key_hash
+      * Attestation.attestation_status)
+      list ->
       (t * History_cache.t) tzresult
 
     type proof
diff --git a/src/proto_alpha/lib_protocol/apply.ml b/src/proto_alpha/lib_protocol/apply.ml
index 94b9f3172fbbea95a34a7bf4de697aa19961908e..93939908588235885aa95ee0422b39decc55134b 100644
--- a/src/proto_alpha/lib_protocol/apply.ml
+++ b/src/proto_alpha/lib_protocol/apply.ml
@@ -1438,7 +1438,7 @@ let apply_manager_operation :
             [] )
     | Dal_publish_commitment slot_header ->
         let*? ctxt, slot_header =
-          Dal_apply.apply_publish_commitment ctxt slot_header
+          Dal_apply.apply_publish_commitment ctxt slot_header ~source
         in
         let consumed_gas = Gas.consumed ~since:ctxt_before_op ~until:ctxt in
         let result =
diff --git a/src/proto_alpha/lib_protocol/dal_apply.ml b/src/proto_alpha/lib_protocol/dal_apply.ml
index c3c5b1aa16fc7e41175c7ae1d0c78142caf6a3e4..b79a4819a2544840c0558d9beaa8ceb066da357f 100644
--- a/src/proto_alpha/lib_protocol/dal_apply.ml
+++ b/src/proto_alpha/lib_protocol/dal_apply.ml
@@ -87,7 +87,7 @@ let apply_attestation ctxt attestation ~power =
 let validate_publish_commitment ctxt _operation =
   assert_dal_feature_enabled ctxt
 
-let apply_publish_commitment ctxt operation =
+let apply_publish_commitment ctxt operation ~source =
   let open Result_syntax in
   let* ctxt = Gas.consume ctxt Dal_costs.cost_Dal_publish_commitment in
   let number_of_slots = Dal.number_of_slots ctxt in
@@ -100,7 +100,7 @@ let apply_publish_commitment ctxt operation =
       ~current_level
       operation
   in
-  let* ctxt = Dal.Slot.register_slot_header ctxt slot_header in
+  let* ctxt = Dal.Slot.register_slot_header ctxt slot_header ~source in
   return (ctxt, slot_header)
 
 let finalisation ctxt =
diff --git a/src/proto_alpha/lib_protocol/dal_apply.mli b/src/proto_alpha/lib_protocol/dal_apply.mli
index 55bd78051f1a89805bdb6188d641cdd75e68b9a1..138e9b99821445bc13154532a4072c23bdbd8db7 100644
--- a/src/proto_alpha/lib_protocol/dal_apply.mli
+++ b/src/proto_alpha/lib_protocol/dal_apply.mli
@@ -57,11 +57,14 @@ val apply_attestation : t -> Dal.Attestation.t -> power:int -> t tzresult
 val validate_publish_commitment :
   t -> Dal.Operations.Publish_commitment.t -> unit tzresult
 
-(** [apply_publish_commitment ctxt slot_header] applies the publication of
-   slot header [slot_header] on top of [ctxt]. Fails if the slot contains
-   already a slot header. *)
+(** [apply_publish_commitment ctxt slot_header ~source] applies the publication
+    of slot header [slot_header] signed by [source] on top of [ctxt]. Fails if
+    the slot already contains a slot header. *)
 val apply_publish_commitment :
-  t -> Dal.Operations.Publish_commitment.t -> (t * Dal.Slot.Header.t) tzresult
+  t ->
+  Dal.Operations.Publish_commitment.t ->
+  source:public_key_hash ->
+  (t * Dal.Slot.Header.t) tzresult
 
 (** [finalisation ctxt] should be executed at block finalisation
    time. A set of slots attested at level [ctxt.current_level - lag]
diff --git a/src/proto_alpha/lib_protocol/dal_slot_repr.ml b/src/proto_alpha/lib_protocol/dal_slot_repr.ml
index 87fc6d51ab7e0dca6deb3809b4f3cec5983f77dd..977758c6110b93542621965f893de75b240eb838 100644
--- a/src/proto_alpha/lib_protocol/dal_slot_repr.ml
+++ b/src/proto_alpha/lib_protocol/dal_slot_repr.ml
@@ -218,7 +218,10 @@ module Slot_market = struct
 
   module Slot_index_map = Map.Make (Dal_slot_index_repr)
 
-  type t = {length : int; slot_headers : Header.t Slot_index_map.t}
+  type t = {
+    length : int;
+    slot_headers : (Header.t * Signature.public_key_hash) Slot_index_map.t;
+  }
 
   let init ~length =
     if Compare.Int.(length < 0) then
@@ -228,7 +231,7 @@ module Slot_market = struct
 
   let length {length; _} = length
 
-  let register t new_slot_header =
+  let register t new_slot_header ~source =
     let open Header in
     if
       not
@@ -241,7 +244,7 @@ module Slot_market = struct
       let update = function
         | None ->
             has_changed := true ;
-            Some new_slot_header
+            Some (new_slot_header, source)
         | Some x -> Some x
       in
       let slot_headers =
@@ -310,7 +313,7 @@ module History = struct
         | _ -> None)
       (fun () -> Add_element_in_slots_skip_list_violates_ordering)
 
-  module Content = struct
+  module Content_v1 = struct
     (** Each cell of the skip list is either a slot header that has been
         attested, or a published level and a slot index for which no slot header
         is attested (so, no associated commitment). *)
@@ -362,9 +365,237 @@ module History = struct
           Format.fprintf fmt "Unattested (%a)" Header.pp_id slot_id
       | Attested slot_header ->
           Format.fprintf fmt "Attested (%a)" Header.pp slot_header
+
+    let to_bytes current_slot =
+      Data_encoding.Binary.to_bytes_exn encoding current_slot
   end
 
-  module Skip_list = struct
+  module Content_v2 = struct
+    [@@@warning "-32"]
+
+    (** Each cell of the skip list is either a slot id (i.e. a published level
+        and a slot index) for which no slot header is published or a published
+        slot header associated to the address which signed the L1 operation, the
+        attestation status from the protocol point of view, the number of
+        attested shards and the total number of shards. *)
+    type t =
+      | Unpublished of Header.id
+      | Published of {
+          header : Header.t;
+          publisher : Signature.public_key_hash;
+          is_proto_attested : bool;
+          attested_shards : int;
+          total_shards : int;
+        }
+
+    let content_id = function
+      | Unpublished slot_id -> slot_id
+      | Published {header = {id; _}; _} -> id
+
+    let encoding =
+      let open Data_encoding in
+      (* ADAL/TODO: https://gitlab.com/tezos/tezos/-/issues/7554
+
+         The two legacy cases below can be removed once migration from Quebec to
+         R is done, assuming no legacy cells could be used in refutation anymore
+         (i.e. ~3 months after migrating to Content_v2). If so, the code in
+         protocol S can be simplfied.
+
+         However, make sure to not modify the tags (2) and (3) of the regular
+         cases below, as this would break the ability to read normal Content_v2
+         cells. *)
+      union
+        ~tag_size:`Uint8
+        [
+          (* Legacy cases: we should be able to decode the content of the old
+             cells from the previous protocol in case of refutation. But, we
+             don't construct cells of that protocol anymore. *)
+          case
+            ~title:"unattested"
+            (Tag 0)
+            (merge_objs
+               (obj1 (req "kind" (constant "unattested")))
+               Header.id_encoding)
+            (fun _x -> None)
+            (fun ((), slot_id) ->
+              (* The old [Unattested] case is translated to [Unpublished]. When
+                 a slot was not attested, we act as if it was not published at
+                 all. *)
+              Unpublished slot_id);
+          case
+            ~title:"attested"
+            (Tag 1)
+            (merge_objs
+               (obj1 (req "kind" (constant "attested")))
+               Header.encoding)
+            (fun _x -> None)
+            (fun ((), slot_header) ->
+              (* The old [Attested] case is translated to [Published], with some
+                 extra placeholder fields. In this case, we act as if it were
+                 attested by all the bakers (i.e., a 100% attestation rate). *)
+              Published
+                {
+                  header = slot_header;
+                  publisher = Signature.Public_key_hash.zero;
+                  is_proto_attested = true;
+                  attested_shards = 1;
+                  total_shards = 1;
+                });
+          (* New/normal cases *)
+          case
+            ~title:"unpublished"
+            (Tag 2)
+            (merge_objs
+               (obj1 (req "kind" (constant "unpublished")))
+               Header.id_encoding)
+            (function
+              | Unpublished slot_id -> Some ((), slot_id) | Published _ -> None)
+            (fun ((), slot_id) -> Unpublished slot_id);
+          case
+            ~title:"published"
+            (Tag 3)
+            (merge_objs
+               (obj5
+                  (req "kind" (constant "published"))
+                  (req "publisher" Signature.Public_key_hash.encoding)
+                  (req "is_proto_attested" bool)
+                  (req "attested_shards" uint16)
+                  (req "total_shards" uint16))
+               Header.encoding)
+            (function
+              | Unpublished _ -> None
+              | Published
+                  {
+                    header;
+                    publisher;
+                    is_proto_attested;
+                    attested_shards;
+                    total_shards;
+                  } ->
+                  Some
+                    ( ( (),
+                        publisher,
+                        is_proto_attested,
+                        attested_shards,
+                        total_shards ),
+                      header ))
+            (fun ( ( (),
+                     publisher,
+                     is_proto_attested,
+                     attested_shards,
+                     total_shards ),
+                   header ) ->
+              Published
+                {
+                  header;
+                  publisher;
+                  is_proto_attested;
+                  attested_shards;
+                  total_shards;
+                });
+        ]
+
+    let equal t1 t2 =
+      match (t1, t2) with
+      | Unpublished sid1, Unpublished sid2 -> Header.slot_id_equal sid1 sid2
+      | ( Published
+            {
+              header;
+              publisher;
+              is_proto_attested;
+              attested_shards;
+              total_shards;
+            },
+          Published sh ) ->
+          Header.equal header sh.header
+          && Signature.Public_key_hash.equal publisher sh.publisher
+          && Compare.Bool.equal is_proto_attested sh.is_proto_attested
+          && Compare.Int.equal attested_shards sh.attested_shards
+          && Compare.Int.equal total_shards sh.total_shards
+      | Unpublished _, _ | Published _, _ -> false
+
+    let zero, zero_level =
+      let zero_level = Raw_level_repr.root in
+      let zero_index = Dal_slot_index_repr.zero in
+      ( Unpublished {published_level = zero_level; index = zero_index},
+        zero_level )
+
+    let pp fmt = function
+      | Unpublished slot_id ->
+          Format.fprintf fmt "Unpublished (%a)" Header.pp_id slot_id
+      | Published
+          {header; publisher; is_proto_attested; attested_shards; total_shards}
+        ->
+          Format.fprintf
+            fmt
+            "Published { @[header: %a@@] @[publisher: %a@@] \
+             @[is_proto_attested: %b@@] @[attested_shards: %d@@] \
+             @[total_shards: %d@@] }"
+            Header.pp
+            header
+            Signature.Public_key_hash.pp
+            publisher
+            is_proto_attested
+            attested_shards
+            total_shards
+
+    (* ADAL/TODO: https://gitlab.com/tezos/tezos/-/issues/7554
+
+       The implementation of [to_bytes] can be simplified after migration. See the
+       similar TODO in the encoding above for more details.
+    *)
+    let to_bytes =
+      let to_new_bytes current_slot =
+        Data_encoding.Binary.to_bytes_exn encoding current_slot
+      in
+      (* This function, and the [hash] function when {!Content_v2} are now
+         parameterized by an optional value [with_migration] for the following
+         reason:
+
+         Outside of refutation games, [with_migration] is set to
+         [None]. However, when verifying a proof (for proofs production, the
+         hashes are provided, not computed), we need to hash the cells along the
+         path from the snapshot cell to the target cell.
+
+         Hashing the cells of the previous protocol should be done using the
+         previous representation/encoding. To be able to do so, the proof
+         validation function should take a parameter [with_migration], which
+         includes the attestation lag and the level at which the migration
+         occurred. This way, we'll able to determine in which protocol we hashed
+         the content the first time (and computed a backpointer from the hash)
+         to continue using the same hash function on the same representation. *)
+      fun ?with_migration current_slot ->
+        match with_migration with
+        | None -> to_new_bytes current_slot
+        | Some (migration_level, attestation_lag) -> (
+            let Header.{published_level; _} = content_id current_slot in
+            let attested_level =
+              Raw_level_repr.add published_level attestation_lag
+            in
+            if Raw_level_repr.(attested_level >= migration_level) then
+              (* If attested_level is at least equal to migration_level, this
+                 means that the content has always been hashed using the new
+                 encoding. So, we continue doing so. *)
+              to_new_bytes current_slot
+            else
+              (* if attested_level < migration_level, this means that this
+                 content has already been hashed with the representation of
+                 cells' content in the previous protocol. To keep getting the
+                 same hash used as a backpointer, we rehash using the same
+                 (legacy) function. *)
+              Content_v1.to_bytes
+              @@
+              match current_slot with
+              | Unpublished id -> Content_v1.Unattested id
+              | Published {header; _} -> Content_v1.Attested header)
+  end
+
+  module Mk_skip_list (Content : sig
+    type t
+
+    val content_id : t -> Header.id
+  end) =
+  struct
     include Skip_list.Make (Skip_list_parameters)
 
     (** All Dal slot indices for all levels will be stored in a skip list
@@ -427,6 +658,9 @@ module History = struct
   end
 
   module V1 = struct
+    module Content = Content_v1
+    module Skip_list = Mk_skip_list (Content)
+
     type content = Content.t
 
     (* A pointer to a cell is the hash of its content and all the back
@@ -482,7 +716,7 @@ module History = struct
     let hash cell =
       let current_slot = Skip_list.content cell in
       let back_pointers_hashes = Skip_list.back_pointers cell in
-      Data_encoding.Binary.to_bytes_exn Content.encoding current_slot
+      Content.to_bytes current_slot
       :: List.map Pointer_hash.to_bytes back_pointers_hashes
       |> Pointer_hash.hash_bytes
 
@@ -561,8 +795,11 @@ module History = struct
       let mk_unattested index =
         Content.Unattested Header.{published_level; index}
       in
-      (* TODO: Follow-up MR: Take the value of _s_status into account. *)
-      let attested_slot_headers = List.map fst slot_headers_with_statuses in
+      (* TODO: Follow-up MR: Take the value of _s_status and _s_publisher into
+         account. *)
+      let attested_slot_headers =
+        List.map (fun (slot, _pub, _status) -> slot) slot_headers_with_statuses
+      in
       (* Hypothesis: both lists are sorted in increasing order w.r.t. slots
          indices. *)
       let rec aux indices slots =
@@ -591,7 +828,7 @@ module History = struct
       let open Result_syntax in
       let* () =
         List.iter_e
-          (fun (slot_header, _status) ->
+          (fun (slot_header, _slot_publisher, _status) ->
             error_unless
               Raw_level_repr.(
                 published_level = slot_header.Header.id.published_level)
diff --git a/src/proto_alpha/lib_protocol/dal_slot_repr.mli b/src/proto_alpha/lib_protocol/dal_slot_repr.mli
index 568e27b34844b7bc48092dc710d596c08821e193..09c655731a35d9cd8bf603a64aeabdec56d3b6b0 100644
--- a/src/proto_alpha/lib_protocol/dal_slot_repr.mli
+++ b/src/proto_alpha/lib_protocol/dal_slot_repr.mli
@@ -187,15 +187,18 @@ module Slot_market : sig
      (see {!val:init}). *)
   val length : t -> int
 
-  (** [register t index fees] updates the candidate associated to
-     index [index]. Returns [Some (_, true)] if the candidate is
-     registered. Returns [Some (_, false)] otherwise. Returns [None]
-     if the [index] is not in the interval [0;length] where [length]
-     is the value provided to the [init] function. *)
-  val register : t -> Header.t -> (t * bool) option
-
-  (** [candidates t] returns a list of slot header candidates. *)
-  val candidates : t -> Header.t list
+  (** [register t slot_header ~source] possibly updates the candidate associated
+      to [slot_header.id] with the [slot_header] published by [source].
+
+      The function returns [Some (_, true)] if the candidate is
+      registered. Returns [Some (_, false)] otherwise. Returns [None] if
+      [slot_header.id] is not a valid slot id. *)
+  val register :
+    t -> Header.t -> source:Signature.public_key_hash -> (t * bool) option
+
+  (** [candidates t] returns a list of slot header candidates associated to the
+      public key hashes of the managers who published them. *)
+  val candidates : t -> (Header.t * Signature.public_key_hash) list
 end
 
 (** This module provides an abstract data structure (type {!History.t}) that
@@ -275,7 +278,10 @@ module History : sig
     History_cache.t ->
     Raw_level_repr.t ->
     number_of_slots:int ->
-    (Header.t * Dal_attestation_repr.Accountability.attestation_status) list ->
+    (Header.t
+    * Signature.public_key_hash
+    * Dal_attestation_repr.Accountability.attestation_status)
+    list ->
     (t * History_cache.t) tzresult
 
   (** Similiar to {!update_skip_list}, but no cache is provided or
@@ -284,7 +290,10 @@ module History : sig
     t ->
     Raw_level_repr.t ->
     number_of_slots:int ->
-    (Header.t * Dal_attestation_repr.Accountability.attestation_status) list ->
+    (Header.t
+    * Signature.public_key_hash
+    * Dal_attestation_repr.Accountability.attestation_status)
+    list ->
     t tzresult
 
   (** [equal a b] returns true iff a is equal to b. *)
diff --git a/src/proto_alpha/lib_protocol/dal_slot_storage.ml b/src/proto_alpha/lib_protocol/dal_slot_storage.ml
index a5f8621c6bf4405e0b3f43c03066f1d26acb10ef..ebacce2747c61cc9abd90a90ffe9dc004ddf5db4 100644
--- a/src/proto_alpha/lib_protocol/dal_slot_storage.ml
+++ b/src/proto_alpha/lib_protocol/dal_slot_storage.ml
@@ -35,10 +35,11 @@ let finalize_current_slot_headers ctxt =
 
 let compute_slot_headers_statuses ~is_slot_attested seen_slot_headers =
   let open Dal_slot_repr in
-  let fold_attested_slots (rev_attested_slot_headers, attestation) slot =
+  let fold_attested_slots (rev_attested_slot_headers, attestation)
+      (slot, slot_publisher) =
     let attestation_status = is_slot_attested slot in
     let rev_attested_slot_headers =
-      (slot, attestation_status) :: rev_attested_slot_headers
+      (slot, slot_publisher, attestation_status) :: rev_attested_slot_headers
     in
     let attestation =
       if
@@ -96,6 +97,11 @@ let finalize_pending_slot_headers ctxt ~number_of_slots =
   match Raw_level_repr.(sub raw_level dal.attestation_lag) with
   | None -> return (ctxt, Dal_attestation_repr.empty)
   | Some level_attested ->
+      (* TODO/DAL: https://gitlab.com/tezos/tezos/-/issues/7563
+
+         rename a level_attested into a published_level. The name level_attested
+         is misteading as it's equal to (current_)level - attestation_lag, which
+         rather corresponds to the publication level of slot headers. *)
       let* seen_slots = find_slot_headers ctxt level_attested in
       let*! ctxt = Storage.Dal.Slot.Headers.remove ctxt level_attested in
       let* ctxt, attestation, slot_headers_statuses =
diff --git a/src/proto_alpha/lib_protocol/dal_slot_storage.mli b/src/proto_alpha/lib_protocol/dal_slot_storage.mli
index d93cb76463eb8a08c1660c7f5e8decaabdf0343b..746520a82800eb07360c920f869dc79570185c68 100644
--- a/src/proto_alpha/lib_protocol/dal_slot_storage.mli
+++ b/src/proto_alpha/lib_protocol/dal_slot_storage.mli
@@ -44,14 +44,16 @@
     - For every level below [current_level - attestation_lag], there should not be
    any slot in the storage.  *)
 
-(** [find_slot_headers ctxt level] returns [Some slot_headers] where [slot_headers]
-   are pending slots at level [level].  [None] is returned if no
-   [slot_header] was registered at this level. The function fails if
-   the reading into the context fails. *)
+(** [find_slot_headers ctxt level] returns [Some slot_headers] where
+    [slot_headers] are pending slots at level [level] alongside the public key
+    hashes that published them.  [None] is returned if no [slot_header] was
+    registered at this level. The function fails if the reading into the context
+    fails. *)
 val find_slot_headers :
   Raw_context.t ->
   Raw_level_repr.t ->
-  Dal_slot_repr.Header.t list option tzresult Lwt.t
+  (Dal_slot_repr.Header.t * Signature.public_key_hash) list option tzresult
+  Lwt.t
 
 (** [find_level_histories ctxt] returns the cells of the DAL skip list produced
     at the current level alongside their hashes. Returns [None] if no entry is
diff --git a/src/proto_alpha/lib_protocol/init_storage.ml b/src/proto_alpha/lib_protocol/init_storage.ml
index 9b04c21d56e7430caaba86cb529920550e4b726e..af191f5656bbd5ea5f9260aedbb622a95602f3e7 100644
--- a/src/proto_alpha/lib_protocol/init_storage.ml
+++ b/src/proto_alpha/lib_protocol/init_storage.ml
@@ -109,6 +109,37 @@ let patch_script ctxt (address, hash, patched_code) =
         address ;
       return ctxt
 
+(* This temporary function is meant to be used when migrating from Quebec to R.
+   It removes possible entries from the [Storage.Dal.Slot.Headers] map, as the
+   type of values in this map evolved. *)
+let remove_old_published_dal_slot_headers ctxt ~level ~attestation_lag =
+  (* DAL/TODO: https://gitlab.com/tezos/tezos/-/issues/7562
+
+     Add a test that triggers this migration:
+     - Cover all corner cases
+     - Check that all data have been correctly wiped.
+  *)
+  let open Lwt_syntax in
+  (* This function removes all entries at levels [LVL] from
+     [Storage.Dal.Slot.Headers], where:
+
+     max (0, level - attestation_lag) <= LVL <= level
+  *)
+  let rec aux ctxt n =
+    if Compare.Int.(n > attestation_lag) then
+      (* All entries have been wiped. *)
+      Lwt.return ctxt
+    else
+      match Raw_level_repr.sub level n with
+      | None ->
+          (* This happens when level < attestation_lag *)
+          Lwt.return ctxt
+      | Some pub_level ->
+          let* ctxt = Storage.Dal.Slot.Headers.remove ctxt pub_level in
+          aux ctxt (n + 1)
+  in
+  aux ctxt 0
+
 let prepare_first_block chain_id ctxt ~typecheck_smart_contract
     ~typecheck_smart_rollup ~level ~timestamp ~predecessor =
   let open Lwt_result_syntax in
@@ -215,6 +246,12 @@ let prepare_first_block chain_id ctxt ~typecheck_smart_contract
         let* ctxt =
           Sc_rollup_refutation_storage.migrate_clean_refutation_games ctxt
         in
+        let*! ctxt =
+          remove_old_published_dal_slot_headers
+            ctxt
+            ~level
+            ~attestation_lag:parametric.dal.attestation_lag
+        in
         return (ctxt, [])
     (* End of alpha predecessor stitching. Comment used for automatic snapshot *)
   in
diff --git a/src/proto_alpha/lib_protocol/raw_context.ml b/src/proto_alpha/lib_protocol/raw_context.ml
index fa494e315d3ee0c780194860953d2e4bb07b744d..2abb8a55ff979de7d323ffb0fc6db6685e832b17 100644
--- a/src/proto_alpha/lib_protocol/raw_context.ml
+++ b/src/proto_alpha/lib_protocol/raw_context.ml
@@ -2144,12 +2144,13 @@ module Dal = struct
     in
     {ctxt with back = {ctxt.back with dal_attestation_slot_accountability}}
 
-  let register_slot_header ctxt slot_header =
+  let register_slot_header ctxt slot_header ~source =
     let open Result_syntax in
     match
       Dal_slot_repr.Slot_market.register
         ctxt.back.dal_slot_fee_market
         slot_header
+        ~source
     with
     | None ->
         let length =
diff --git a/src/proto_alpha/lib_protocol/raw_context.mli b/src/proto_alpha/lib_protocol/raw_context.mli
index d9de5caa53419c236e2bc5f1816125e7977d65ee..deb8d9ce56e2a56d3de0578b6392116af0ce062c 100644
--- a/src/proto_alpha/lib_protocol/raw_context.mli
+++ b/src/proto_alpha/lib_protocol/raw_context.mli
@@ -441,17 +441,21 @@ module Dal : sig
       available by some attester). *)
   val record_number_of_attested_shards : t -> Dal_attestation_repr.t -> int -> t
 
-  (** [register_slot_header ctxt slot_header] returns a new context
-     where the new candidate [slot] have been taken into
-     account. Returns [Some (ctxt,updated)] where [updated=true] if
-     the candidate is registered. [Some (ctxt,false)] if another
-     candidate was already registered previously. Returns an error if
-     the slot is invalid. *)
-  val register_slot_header : t -> Dal_slot_repr.Header.t -> t tzresult
-
-  (** [candidates ctxt] returns the current list of slot for which
-     there is at least one candidate. *)
-  val candidates : t -> Dal_slot_repr.Header.t list
+  (** [register_slot_header ctxt slot_header ~source] returns a new context
+      where the new candidate [slot] published by [source] has been taken into
+      account. Returns [Some (ctxt,updated)] where [updated=true] if the
+      candidate is registered. [Some (ctxt,false)] if another candidate was
+      already registered previously. Returns an error if the slot is invalid. *)
+  val register_slot_header :
+    t ->
+    Dal_slot_repr.Header.t ->
+    source:Signature.public_key_hash ->
+    t tzresult
+
+  (** [candidates ctxt] returns the current list of slot for which there is at
+      least one candidate alongside the addresses that published them. *)
+  val candidates :
+    t -> (Dal_slot_repr.Header.t * Signature.public_key_hash) list
 
   (** [is_slot_index_attested ctxt slot_index] returns [true] if the
       [slot_index] is declared available by the protocol. [false] otherwise. If
diff --git a/src/proto_alpha/lib_protocol/storage.ml b/src/proto_alpha/lib_protocol/storage.ml
index 8d295e9b4de1be38a9c87733ed2d3a90ebe685fa..6910691f62d58e08f51d1f39cd44377fc3fb6b49 100644
--- a/src/proto_alpha/lib_protocol/storage.ml
+++ b/src/proto_alpha/lib_protocol/storage.ml
@@ -2181,9 +2181,14 @@ module Dal = struct
           (* The size of the list below is at most equal to the
              [number_of_slots] as declared in the DAL parameters of the
              protocol. *)
-          type t = Dal_slot_repr.Header.t list
+          type t = (Dal_slot_repr.Header.t * Signature.public_key_hash) list
 
-          let encoding = Data_encoding.(list Dal_slot_repr.Header.encoding)
+          let encoding =
+            let open Data_encoding in
+            list
+              (obj2
+                 (req "slot_header" Dal_slot_repr.Header.encoding)
+                 (req "publisher" Signature.Public_key_hash.encoding))
         end)
 
     module History =
diff --git a/src/proto_alpha/lib_protocol/storage.mli b/src/proto_alpha/lib_protocol/storage.mli
index 9250bbf2c8ed93d2916596798791ebc43649424c..26e919be8848d114c7a95e2ea9894c5ffb69815d 100644
--- a/src/proto_alpha/lib_protocol/storage.mli
+++ b/src/proto_alpha/lib_protocol/storage.mli
@@ -1015,7 +1015,8 @@ module Dal : sig
       Non_iterable_indexed_data_storage
         with type t = Raw_context.t
          and type key = Raw_level_repr.t
-         and type value = Dal_slot_repr.Header.t list
+         and type value =
+          (Dal_slot_repr.Header.t * Signature.public_key_hash) list
 
     (** This is a permanent storage for slot headers confirmed by the L1. *)
     module History :
diff --git a/src/proto_alpha/lib_protocol/test/pbt/test_dal_slot_proof.ml b/src/proto_alpha/lib_protocol/test/pbt/test_dal_slot_proof.ml
index d3808cc558b1dc95d9a317c73e6bf3c038577ceb..b3f9781c035de802c0a14cc63d4a37fedacb3e4d 100644
--- a/src/proto_alpha/lib_protocol/test/pbt/test_dal_slot_proof.ml
+++ b/src/proto_alpha/lib_protocol/test/pbt/test_dal_slot_proof.ml
@@ -53,8 +53,9 @@ struct
 
   (* Introduce some intermediate types. *)
 
-  (** The slot is not confirmed (skipped) iff the boolean is [true]. *)
-  type slot_skipped = bool
+  (** The slot is not confirmed (skipped) iff the boolean is [true]. The flag is
+      attached to the hypothetical publisher of the slot. *)
+  type slot_skipped = bool * Signature.public_key_hash
 
   type level = int
 
@@ -86,7 +87,7 @@ struct
       let curr_level = Raw_level_repr.of_int32_exn (Int32.of_int level) in
       let slots_headers =
         List.mapi
-          (fun sindex skip_slot ->
+          (fun sindex (skip_slot, slot_publisher) ->
             let index =
               Option.value_f
                 (Dal_slot_index_repr.of_int_opt
@@ -96,12 +97,13 @@ struct
             in
             ( Dal_slot_repr.Header.
                 {id = {published_level = curr_level; index}; commitment},
+              slot_publisher,
               skip_slot ))
           slots_data
       in
       let slots_headers =
         List.filter_map
-          (fun (slot, skip_slot) ->
+          (fun (slot, slot_publisher, skip_slot) ->
             let attestation_status =
               Dal_attestation_repr.Accountability.
                 {
@@ -110,7 +112,8 @@ struct
                   is_proto_attested = not skip_slot;
                 }
             in
-            if skip_slot then None else Some (slot, attestation_status))
+            if skip_slot then None
+            else Some (slot, slot_publisher, attestation_status))
           slots_headers
       in
       let*?@ cell, cache =
@@ -123,8 +126,8 @@ struct
       in
       let slots_info =
         List.fold_left
-          (fun slots_info (slot, slot_status) ->
-            (polynomial, slot, slot_status) :: slots_info)
+          (fun slots_info (slot, slot_publisher, slot_status) ->
+            (polynomial, slot, slot_publisher, slot_status) :: slots_info)
           slots_info
           slots_headers
       in
@@ -137,7 +140,7 @@ struct
 
   (** This function returns the (correct) information of a page to
       prove that it is confirmed, or None if the page's slot is skipped. *)
-  let request_confirmed_page (poly, slot, slot_status) =
+  let request_confirmed_page (poly, slot, _slot_publisher, slot_status) =
     let open Lwt_result_syntax in
     if not slot_status.Dal_attestation_repr.Accountability.is_proto_attested
     then
@@ -152,7 +155,7 @@ struct
       (but the slot is not confirmed). Otherwise, we increment the publish_level
       field to simulate a non confirmed slot (as for even levels, no slot is
       confirmed. See {!populate_slots_history}). *)
-  let request_unconfirmed_page (poly, slot, slot_status) =
+  let request_unconfirmed_page (poly, slot, _slot_publisher, slot_status) =
     let open Lwt_result_syntax in
     let open Dal_slot_repr.Header in
     if not slot_status.Dal_attestation_repr.Accountability.is_proto_attested
@@ -215,7 +218,12 @@ struct
 
   let tests =
     let gen_dal_config : levels QCheck2.Gen.t =
-      QCheck2.Gen.(
+      let open QCheck2 in
+      let gen_pkh =
+        let pkh, _, _ = Signature.generate_key ~algo:Ed25519 () in
+        Gen.return pkh
+      in
+      Gen.(
         let nb_slots = 0 -- Parameters.(dal_parameters.number_of_slots) in
         let nb_levels = 4 -- 30 in
         let* start_level =
@@ -224,7 +232,11 @@ struct
           (* skip level 0 by adding 1 *)
         in
         (* The slot is confirmed iff the boolean is true *)
-        let slot = bool in
+        let slot =
+          let* slot_flag = bool in
+          let* slot_publisher = gen_pkh in
+          Gen.return (slot_flag, slot_publisher)
+        in
         let slots = list_size nb_slots slot in
         (* For each level, we generate the gap/delta w.r.t. the previous level,
            and the slots' flags (confirmed or not). *)
diff --git a/src/proto_alpha/lib_protocol/test/pbt/test_sc_rollup_encoding.ml b/src/proto_alpha/lib_protocol/test/pbt/test_sc_rollup_encoding.ml
index e24ca6075133c6182cc866e40fce31f55ccef13f..753a63085830f450c06c9c7f1881e9716b33a5c6 100644
--- a/src/proto_alpha/lib_protocol/test/pbt/test_sc_rollup_encoding.ml
+++ b/src/proto_alpha/lib_protocol/test/pbt/test_sc_rollup_encoding.ml
@@ -124,16 +124,19 @@ let pack_slots_headers_by_level list =
   let module SSH = Set.Make (struct
     type t =
       Dal_slot_repr.Header.t
+      * Signature.public_key_hash
       * Dal_attestation_repr.Accountability.attestation_status
 
-    let compare (a, _) (b, _) =
+    let compare (a, _, _) (b, _, _) =
       let open Dal_slot_repr.Header in
       Dal_slot_index_repr.compare a.id.index b.id.index
   end) in
   let map =
     List.fold_left
       (fun map
-           ((Dal_slot_repr.Header.{id = {published_level; _}; _}, _status) as sh) ->
+           (( Dal_slot_repr.Header.{id = {published_level; _}; _},
+              _publisher,
+              _status ) as sh) ->
         let l =
           ML.find published_level map |> Option.value ~default:SSH.empty
         in
@@ -157,6 +160,10 @@ let pack_slots_headers_by_level list =
       |> ML.bindings
       |> List.map (fun (k, ssh) -> (k, SSH.elements ssh))
 
+let gen_pkh =
+  let pkh, _, _ = Signature.generate_key ~algo:Ed25519 () in
+  Gen.return pkh
+
 let gen_dal_slots_history () =
   let open Gen in
   let open Dal_slot_repr in
@@ -164,11 +171,12 @@ let gen_dal_slots_history () =
     Tezos_protocol_alpha_parameters.Default_parameters.constants_test
   in
   let number_of_slots = constants.dal.number_of_slots in
-  (* Generate a list of (level * confirmed slot ID). *)
-  let* list = small_list (triple small_nat small_nat bool) in
+  (* Generate a list of (level * confirmed slot ID * public key hash *
+     attestation flag). *)
+  let* list = small_list (quad small_nat small_nat gen_pkh bool) in
   let list =
     List.rev_map
-      (fun (level, slot_index, is_proto_attested) ->
+      (fun (level, slot_index, publisher, is_proto_attested) ->
         let attestation_status =
           Dal_attestation_repr.Accountability.
             {
@@ -188,6 +196,7 @@ let gen_dal_slots_history () =
           |> Option.value ~default:Index.zero
         in
         ( Header.{id = {published_level; index}; commitment = Commitment.zero},
+          publisher,
           attestation_status ))
       list
   in
@@ -197,7 +206,8 @@ let gen_dal_slots_history () =
         let slot_headers =
           (* Sort the list in the right ordering before adding slots to slots_history. *)
           List.sort_uniq
-            (fun ({Header.id = a; _}, _status) ({id = b; _}, _status) ->
+            (fun ({Header.id = a; _}, _publisher, _status)
+                 ({id = b; _}, _publisher, _status) ->
               let c =
                 Raw_level_repr.compare a.published_level b.published_level
               in
diff --git a/src/proto_alpha/lib_protocol/test/unit/test_dal_slot_proof.ml b/src/proto_alpha/lib_protocol/test/unit/test_dal_slot_proof.ml
index f767d21e5971348034c869204eeeb7d9add8937d..734802b9cd3615c5c5c0e831d05657ba324148bf 100644
--- a/src/proto_alpha/lib_protocol/test/unit/test_dal_slot_proof.ml
+++ b/src/proto_alpha/lib_protocol/test/unit/test_dal_slot_proof.ml
@@ -70,7 +70,7 @@ struct
       Hist.update_skip_list_no_cache
         skip_list
         level
-        [(slot, mk_attested)]
+        [(slot, Signature.Public_key_hash.zero, mk_attested)]
         ~number_of_slots:Parameters.dal_parameters.number_of_slots
     in
     check_result result
@@ -196,7 +196,7 @@ struct
         genesis_history
         genesis_history_cache
         level
-        [(slot, mk_attested)]
+        [(slot, Signature.Public_key_hash.zero, mk_attested)]
     in
     let* page_info, page_id = mk_page_info slot polynomial in
     produce_and_verify_proof
diff --git a/src/proto_alpha/lib_sc_rollup_node/test/test_octez_conversions.ml b/src/proto_alpha/lib_sc_rollup_node/test/test_octez_conversions.ml
index 614a3c5d26d234f0d58d330261a60c2448c14c51..2086d6c9519513aa3e4910aa4a0426bdb4392da4 100644
--- a/src/proto_alpha/lib_sc_rollup_node/test/test_octez_conversions.ml
+++ b/src/proto_alpha/lib_sc_rollup_node/test/test_octez_conversions.ml
@@ -229,30 +229,31 @@ let gen_slot_header_with_status =
     let attested_shards = if status then total_shards else 0 in
     Dal.Attestation.{total_shards; attested_shards; is_proto_attested = status}
   in
+  let* publisher = gen_pkh in
   let+ header = gen_slot_header in
-  (header, status)
+  (header, publisher, status)
 
 let gen_slot_headers_with_statuses =
   let open QCheck2.Gen in
   let size = int_bound 50 in
   let+ l = list_size size gen_slot_header_with_status in
   List.sort
-    (fun ((h1 : Octez_smart_rollup.Dal.Slot_header.t), _status1)
-         ((h2 : Octez_smart_rollup.Dal.Slot_header.t), _status2) ->
+    (fun ((h1 : Octez_smart_rollup.Dal.Slot_header.t), _publisher, _status1)
+         ((h2 : Octez_smart_rollup.Dal.Slot_header.t), _publisher, _status2) ->
       compare_slot_header_id h1.id h2.id)
     l
   |> fun l ->
   match l with
   | [] -> []
-  | ((h : Octez_smart_rollup.Dal.Slot_header.t), _status) :: _ ->
+  | ((h : Octez_smart_rollup.Dal.Slot_header.t), _publisher, _status) :: _ ->
       let min_level = h.id.published_level in
       (* smallest level *)
       List.mapi
-        (fun i ((h : Octez_smart_rollup.Dal.Slot_header.t), status) ->
+        (fun i ((h : Octez_smart_rollup.Dal.Slot_header.t), publisher, status) ->
           (* patch the published level to comply with the invariants *)
           let published_level = Int32.(add min_level (of_int i)) in
           let h = {h with id = {h.id with published_level}} in
-          (published_level, [(h, status)]))
+          (published_level, [(h, publisher, status)]))
         l
 
 let gen_slot_history =
@@ -264,8 +265,9 @@ let gen_slot_history =
       (fun (lvl, h) ->
         ( Raw_level.of_int32_exn lvl,
           List.map
-            (fun (h, status) ->
+            (fun (h, publisher, status) ->
               ( Sc_rollup_proto_types.Dal.Slot_header.of_octez ~number_of_slots h,
+                publisher,
                 status ))
             h ))
       l
@@ -294,8 +296,9 @@ let gen_slot_history_cache =
       (fun (lvl, h) ->
         ( Raw_level.of_int32_exn lvl,
           List.map
-            (fun (h, status) ->
+            (fun (h, publisher, status) ->
               ( Sc_rollup_proto_types.Dal.Slot_header.of_octez ~number_of_slots h,
+                publisher,
                 status ))
             h ))
       l