Merge branch 'rightlib' into merge-libs-241228-0020

53 files changed, 512 insertions, 445 deletions

diff --git a/build/conf/docs.conf b/build/conf/docs.conf
index 0cd8c7d20db..b622013c7aa 100644
--- a/build/conf/docs.conf
+++ b/build/conf/docs.conf
@@ -125,7 +125,7 @@ module _DOCS_BASE_UNIT: _DOCS_BARE_UNIT {
 }
 
 # tag:internal tag:docs
-### _DOCS_YFM_USE_PLANTUML() # internal
+### @usage: _DOCS_YFM_USE_PLANTUML() # internal
 ###
 ### This macr sets appropriate dependencies for use of plantuml plugin
 macro _DOCS_YFM_USE_PLANTUML() {
diff --git a/build/conf/java.conf b/build/conf/java.conf
index c83a5835260..fc846dfdac8 100644
--- a/build/conf/java.conf
+++ b/build/conf/java.conf
@@ -126,7 +126,7 @@ multimodule JAVA_CONTRIB_ANNOTATION_PROCESSOR {
     }
 }
 
-### Usage JUNIT_TESTS_JAR(path/to/some/peer realname.jar)
+### @usage: JUNIT_TESTS_JAR(path/to/some/peer realname.jar)
 ###
 ### Specifies jar to search test suites and test cases. By default tests are
 ### searched in the jar compild by JTEST or JUNIT5 module sources. This macro
@@ -1797,7 +1797,7 @@ macro ANNOTATION_PROCESSOR(Args...) {
 }
 
 EXCLUDE_VALUE=
-### EXCLUDE(prefixes)
+### @usage: EXCLUDE(prefixes)
 ###
 ### Specifies which libraries should be excluded from the classpath.
 macro EXCLUDE(Args...) {
@@ -2066,7 +2066,7 @@ IGNORE_JAVA_DEPENDENCIES_CONFIGURATION=no
 JAVA_DEPENDENCIES_CONFIGURATION_VALUE=
 
 # tag:java-specific
-### @usage JAVA_DEPENDENCIES_CONFIGURATION(Vetos...)
+### @usage: JAVA_DEPENDENCIES_CONFIGURATION(Vetos...)
 ### Validate contrib/java dependencies
 ### Valid arguments
 ### FORBID_DIRECT_PEERDIRS - fail when module have direct PEERDIR (with version) (non-transitive)
@@ -2087,7 +2087,7 @@ JAR_INCLUDE_FILTER_VALUE=
 JAR_EXCLUDE_FILTER_VALUE=
 
 # tag:java-specific
-### @usage:JAR_INCLUDE(Filters...)
+### @usage: JAR_INCLUDE(Filters...)
 ### Filter .jar file content: keep only matched files
 ### * and ** patterns are supported (like JAVA_SRCS)
 macro JAR_INCLUDE(Filters...) {
@@ -2095,7 +2095,7 @@ macro JAR_INCLUDE(Filters...) {
 }
 
 # tag:java-specific
-### @usage:JAR_EXCLUDE(Filters...)
+### @usage: JAR_EXCLUDE(Filters...)
 ### Filter .jar file content: remove matched files
 ### * and ** patterns are supported (like JAVA_SRCS)
 macro JAR_EXCLUDE(Filters...) {
@@ -2198,7 +2198,7 @@ macro USE_KTLINT_OLD() {
 _KTLINT_BASELINE_FILE=
 
 # tag:kotlin-specific
-### @usage: #deprecated KTLINT_BASELINE_FILE(ktlint-baseline.xml "https://st.yandex-team.ru/REMOVE-BASELINE-1")
+### @usage: KTLINT_BASELINE_FILE(ktlint-baseline.xml "https://st.yandex-team.ru/REMOVE-BASELINE-1") # deprecated
 ### Path to baseline file for ktlint test and ticket to fix all ktlint warnings in file and then remove it
 macro KTLINT_BASELINE_FILE(File, Ticket) {
     SET(_KTLINT_BASELINE_FILE $File)
diff --git a/build/conf/proto.conf b/build/conf/proto.conf
index 5676cccbb2f..b415e0687f1 100644
--- a/build/conf/proto.conf
+++ b/build/conf/proto.conf
@@ -566,7 +566,7 @@ macro GRPC() {
 _GRPC_GMOCK_OUTFLAG=
 
 # tag:proto tag:grpc
-### @usage: WITH_GMOCK()
+### @usage: GRPC_WITH_GMOCK()
 ###
 ### Enable generating *_mock.grpc.pb.cc/h files
 macro GRPC_WITH_GMOCK() {
diff --git a/build/conf/python.conf b/build/conf/python.conf
index 698f6b9f6ab..8aab5de8ebb 100644
--- a/build/conf/python.conf
+++ b/build/conf/python.conf
@@ -745,7 +745,7 @@ otherwise {
 }
 
 # tag:python-specific
-### @usage: _BASE_PY_PROGRAM #internal
+### @usage: _BASE_PY3_PROGRAM #internal
 ###
 ### The base module for all Python 3.x binary programs. Adds linking logic, relevant module properties and
 ### dependency on Python 3.x interpreter. Also adds import tests on all sources including libraries.
@@ -901,7 +901,7 @@ macro _ARCADIA_PYTHON3_ADDINCL() {
 }
 
 # tag:python-specific  tag:internal
-### @usage: _PYTHON_ADDINCL()  # internal
+### @usage: _PYTHON3_ADDINCL()  # internal
 ###
 ### This macro sets up Python 3.x headers for both Arcadia and non-Arcadia python.
 macro _PYTHON3_ADDINCL() {
diff --git a/build/conf/ts/ts.conf b/build/conf/ts/ts.conf
index f3149b3d3c6..06ba1661dc7 100644
--- a/build/conf/ts/ts.conf
+++ b/build/conf/ts/ts.conf
@@ -121,7 +121,7 @@ macro TS_ESLINT_CONFIG(ConfigName) {
 }
 
 _TS_LINT_SRCS_VALUE=
-### _TS_CONFIG_EPILOGUE() # internal
+### @usage: _TS_CONFIG_EPILOGUE() # internal
 ###
 ### This macro executes macros which should be invoked after all user specified macros in the ya.make file
 macro _TS_CONFIG_EPILOGUE() {
@@ -140,7 +140,7 @@ TSYNDEXER_CMD=${cwd:ARCADIA_BUILD_ROOT} \
     $TSYNDEXER_TOOL -m ${CURDIR} -o ${output:TSYNDEXER_OUTPUT_FILE} -r ${ARCADIA_ROOT} \
     ${hide;kv:"pc magenta"} ${hide;kv:"p YTS"} $_AS_HIDDEN_INPUTS(IN $_TS_LINT_SRCS_VALUE)
 
-### _DO_TS_YNDEXING() # internal
+### @usage: _DO_TS_YNDEXING() # internal
 ###
 ### Adds a command-node for TS-modules codenavigation indexing.
 ### Output ydx.pb2 file will be processed along with other indexes for other modules and
@@ -179,7 +179,7 @@ macro _AS_HIDDEN_INPUTS(IN{input}[]) {
 
 _TS_FILES_COPY_CMD=
 
-### TS_FILES(Files...)
+### @usage: TS_FILES(Files...)
 ###
 ### Adds files to output as is. Similar to FILES but works for TS build modules
 ### Documentation: https://docs.yandex-team.ru/frontend-in-arcadia/references/TS_PACKAGE#ts-files
@@ -188,7 +188,7 @@ macro TS_FILES(Files...) {
     _TS_FILES($Files)
 }
 
-### TS_FILES_GLOB(Glob...)
+### @usage: TS_FILES_GLOB(Glob...)
 ###
 ### Adds files to output by glob, e.g. TS_FILES_GLOB(**/*.css)
 ### Documentation: https://docs.yandex-team.ru/frontend-in-arcadia/references/TS_PACKAGE#ts-files-glob
@@ -197,7 +197,7 @@ macro TS_FILES_GLOB(Glob...) {
     TS_FILES(${FILES_BY_GLOB})
 }
 
-### @usage TS_LARGE_FILES(DESTINATION dest_dir Files...)
+### @usage: TS_LARGE_FILES(DESTINATION dest_dir Files...)
 ###
 ### Use large file ether from working copy or from remote storage via placeholder <File>.external
 ### If <File> is present locally (and not a symlink!) it will be copied to build directory.
diff --git a/build/plugins/extralibs.py b/build/plugins/extralibs.py
index 80f0a70cdc1..1facfa33540 100644
--- a/build/plugins/extralibs.py
+++ b/build/plugins/extralibs.py
@@ -1,6 +1,6 @@
 def onpy_extralibs(unit, *args):
     """
-    @usage: EXTRALIBS(liblist)
+    @usage: PY_EXTRALIBS(liblist)
     Add external dynamic libraries during program linkage stage" }
     """
 
diff --git a/build/ymake.core.conf b/build/ymake.core.conf
index 70b339889c5..76a61790ee9 100644
--- a/build/ymake.core.conf
+++ b/build/ymake.core.conf
@@ -440,7 +440,7 @@ macro CHECK_CONFIG_H(Conf) {
 }
 
 REQUIRED_TRANSITIVE_PEERS=
-### @usage REQUIRES(dirs...)
+### @usage: REQUIRES(dirs...)
 ###
 ### Specify list of dirs which this module must depend on indirectly.
 ###
@@ -452,7 +452,7 @@ macro REQUIRES(DIRS[]) {
 
 CHECK_DEPENDENT_DIRS_TYPES=
 CHECK_DEPENDENT_DIRS_RESTRICTIONS=
-### @usage CHECK_DEPENDENT_DIRS(DENY|ALLOW_ONLY ([ALL|PEERDIRS|GLOB] dir)...)
+### @usage: CHECK_DEPENDENT_DIRS(DENY|ALLOW_ONLY ([ALL|PEERDIRS|GLOB] dir)...)
 ###
 ### Specify project transitive dependencies constraints.
 ###
@@ -479,7 +479,7 @@ macro _RESOURCE_SEM(INPUTS[], KEYS[], OPTS[]) {
 }
 
 # tag:built-in
-### @usage: RESOURCE([FORCE_TEXT ][Src Key]* [- Key=Value]*) # built-in
+### @usage: RESOURCE([FORCE_TEXT ][Src Key]* [- Key=Value]*) # builtin
 ### Add data (resources, random files, strings) to the program)
 ### The common usage is to place Src file into binary. The Key is used to access it using library/cpp/resource or library/python/resource.
 ### Alternative syntax with '- Key=Value' allows placing Value string as resource data into binary and make it accessible by Key.
@@ -1018,7 +1018,7 @@ when ($SO_OUTPUTS == "yes") {
 }
 
 # tag:restricted tag:DLL
-### $usage: WITH_DYNAMIC_LIBS() # restricted
+### @usage: WITH_DYNAMIC_LIBS() # restricted
 ###
 ### Include dynamic libraries as extra PROGRAM/DLL outputs
 macro WITH_DYNAMIC_LIBS() {
@@ -1026,31 +1026,31 @@ macro WITH_DYNAMIC_LIBS() {
 }
 
 # tag:internal
-### $usage: REAL_LINK_EXE_IMPL(peers...) # internal
+### @usage: REAL_LINK_EXE_IMPL(peers...) # internal
 macro REAL_LINK_EXE_IMPL(WHOLE_ARCHIVE_PEERS...) {
     .CMD=$REAL_LINK_EXE_CMDLINE && $LINK_OR_COPY_SO_CMD
 }
 
 # tag:internal
-### $usage: REAL_LINK_EXEC_DYN_LIB_IMPL(peers...) # internal
+### @usage: REAL_LINK_EXEC_DYN_LIB_IMPL(peers...) # internal
 macro REAL_LINK_EXEC_DYN_LIB_IMPL(WHOLE_ARCHIVE_PEERS...) {
     .CMD=$REAL_LINK_EXEC_DYN_LIB_CMDLINE
 }
 
 # tag:internal
-### $usage: REAL_LINK_DYN_LIB_IMPL(peers...) # internal
+### @usage: REAL_LINK_DYN_LIB_IMPL(peers...) # internal
 macro REAL_LINK_DYN_LIB_IMPL(WHOLE_ARCHIVE_PEERS...) {
     .CMD=$REAL_LINK_DYN_LIB_CMDLINE && $LINK_OR_COPY_SO_CMD
 }
 
 # tag:internal
-### $usage: LINK_EXE_IMPL(peers...) # internal
+### @usage: LINK_EXE_IMPL(peers...) # internal
 macro LINK_EXE_IMPL(WHOLE_ARCHIVE_PEERS...) {
     .CMD=$LINK_EXE_CMDLINE && $LINK_OR_COPY_SO_CMD
 }
 
 # tag:internal
-### $usage: LINK_EXEC_DYN_LIB_IMPL(peers...) # internal
+### @usage: LINK_EXEC_DYN_LIB_IMPL(peers...) # internal
 macro LINK_EXEC_DYN_LIB_IMPL(WHOLE_ARCHIVE_PEERS...) {
     .CMD=$LINK_EXEC_DYN_LIB_CMDLINE
 }
@@ -1457,7 +1457,7 @@ module BOOSTTEST: _BASE_PROGRAM {
 }
 
 # tag:deprecated
-### @usage BOOSTTEST_WITH_MAIN([name]) #deprecated
+### @usage: BOOSTTEST_WITH_MAIN([name]) #deprecated
 ###
 ### Same as BOOSTTEST (see above), but comes with builtin int main(argc, argv) implementation
 module BOOSTTEST_WITH_MAIN: BOOSTTEST {
@@ -2578,7 +2578,7 @@ macro EXTRALIBS_STATIC(Args...) {
     LDFLAGS(-Wl,-Bstatic ${Args} -Wl,-Bdynamic)
 }
 
-### @usage ADD_COMPILABLE_TRANSLATE(Dict Name Options...)
+### @usage: ADD_COMPILABLE_TRANSLATE(Dict Name Options...)
 ###
 ### Generate translation dictionary code to transdict.LOWER(Name).cpp that will than be compiled into library
 macro ADD_COMPILABLE_TRANSLATE(Dict, Name, MakeTransDictOptions...) {
@@ -2587,7 +2587,7 @@ macro ADD_COMPILABLE_TRANSLATE(Dict, Name, MakeTransDictOptions...) {
      RUN_PROGRAM(dict/tools/maketransdict -i ${Dict} ${MakeTransDictOptions} ${Name} STDOUT_NOAUTO ${__translate_dict} IN ${Dict})
 }
 
-### @usage ADD_COMPILABLE_TRANSLIT(TranslitTable NGrams Name Options...)
+### @usage: ADD_COMPILABLE_TRANSLIT(TranslitTable NGrams Name Options...)
 ###
 ### Generate transliteration dictionary code
 ### This will emit both translit, untranslit and ngrams table codes those will be than further compiled into library
@@ -2643,7 +2643,7 @@ macro _ARF_HELPER(Args...) {
      RESOURCE_FILES($Args)
 }
 
-### @usage ALL_RESOURCE_FILES(Ext [PREFIX {prefix}] [STRIP {strip}] Dirs...)
+### @usage: ALL_RESOURCE_FILES(Ext [PREFIX {prefix}] [STRIP {strip}] Dirs...)
 ###
 ### This macro collects all files with extension `Ext` and
 ### Passes them to `RESOURCE_FILES` macro as relative to current directory
@@ -2657,7 +2657,7 @@ macro ALL_RESOURCE_FILES(EXT, PREFIX="", STRIP="", DIRS...) {
      _ARF_HELPER(${pre=PREFIX :PREFIX} STRIP ${ARCADIA_ROOT}/${MODDIR}/${STRIP} ${pre=$ALL_RES_:EXT})
 }
 
-### @usage ALL_RESOURCE_FILES_FROM_DIRS([PREFIX {prefix}] [STRIP {strip}] Dirs...)
+### @usage: ALL_RESOURCE_FILES_FROM_DIRS([PREFIX {prefix}] [STRIP {strip}] Dirs...)
 ###
 ### This macro collects all files non-recursively from listed Dirs and
 ### Passes them to `RESOURCE_FILES` macro as relative to current directory
@@ -3435,7 +3435,7 @@ macro _SRC("ydl", SRC, SRCFLAGS...) {
 }
 
 # tag:ydl-specific
-### @usage BUILD_YDL_DESC(Input Symbol Output)
+### @usage: BUILD_YDL_DESC(Input Symbol Output)
 ###
 ### Generate a descriptor for a Symbol located in a ydl module Input, and put it to the file Output.
 ###
@@ -3468,7 +3468,7 @@ macro YDL_DESC_USE_BINARY() {
     SET_APPEND(YDL_DESC_FLAGS $YDL_DESC_FLAGS_BINARY)
 }
 
-### @usage SRC(File Flags...)
+### @usage: SRC(File Flags...)
 ###
 ### Compile single file with extra Flags.
 ### Compilation is driven by the last extension of the File and Flags are specific to corresponding compilation command
@@ -3561,7 +3561,7 @@ macro _SRC_CUSTOM_C_CPP("C", MACRO_NAME, FILE, COMPILE_OUT_SUFFIX, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_SSE2(File Flags...)
+### @usage: SRC_C_SSE2(File Flags...)
 ###
 ### Compile a single C/C++ file with SSE2 and additional Flags
 macro SRC_C_SSE2(FILE, FLAGS...) {
@@ -3569,7 +3569,7 @@ macro SRC_C_SSE2(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_SSE3(File Flags...)
+### @usage: SRC_C_SSE3(File Flags...)
 ###
 ### Compile a single C/C++ file with SSE3 and additional Flags
 macro SRC_C_SSE3(FILE, FLAGS...) {
@@ -3577,7 +3577,7 @@ macro SRC_C_SSE3(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_SSSE3(File Flags...)
+### @usage: SRC_C_SSSE3(File Flags...)
 ###
 ### Compile a single C/C++ file with SSSE3 and additional Flags
 macro SRC_C_SSSE3(FILE, FLAGS...) {
@@ -3585,7 +3585,7 @@ macro SRC_C_SSSE3(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_SSE4(File Flags...)
+### @usage: SRC_C_SSE4(File Flags...)
 ###
 ### Compile a single C/C++ file with SSE4 and additional Flags
 macro SRC_C_SSE4(FILE, FLAGS...) {
@@ -3593,7 +3593,7 @@ macro SRC_C_SSE4(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_SSE41(File Flags...)
+### @usage: SRC_C_SSE41(File Flags...)
 ###
 ### Compile a single C/C++ file with SSE4.1 and additional Flags
 macro SRC_C_SSE41(FILE, FLAGS...) {
@@ -3601,7 +3601,7 @@ macro SRC_C_SSE41(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_AVX(File Flags...)
+### @usage: SRC_C_AVX(File Flags...)
 ###
 ### Compile a single C/C++ file with AVX and additional Flags
 macro SRC_C_AVX(FILE, FLAGS...) {
@@ -3609,7 +3609,7 @@ macro SRC_C_AVX(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_AVX2(File Flags...)
+### @usage: SRC_C_AVX2(File Flags...)
 ###
 ### Compile a single C/C++ file with AVX2 and additional Flags
 macro SRC_C_AVX2(FILE, FLAGS...) {
@@ -3617,7 +3617,7 @@ macro SRC_C_AVX2(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_AVX512(File Flags...)
+### @usage: SRC_C_AVX512(File Flags...)
 ###
 ### Compile a single C/C++ file with AVX512 and additional Flags
 macro SRC_C_AVX512(FILE, FLAGS...) {
@@ -3625,7 +3625,7 @@ macro SRC_C_AVX512(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_AVX512(File Flags...)
+### @usage: SRC_C_AMX(File Flags...)
 ###
 ### Compile a single C/C++ file with AVX512 and additional Flags
 macro SRC_C_AMX(FILE, FLAGS...) {
@@ -3633,7 +3633,7 @@ macro SRC_C_AMX(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_XOP(File Flags...)
+### @usage: SRC_C_XOP(File Flags...)
 ###
 ### Compile a single C/C++ file with (an AMD-specific instruction set,
 ### see https://en.wikipedia.org/wiki/XOP_instruction_set) and additional Flags
@@ -3642,7 +3642,7 @@ macro SRC_C_XOP(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific tag:cpu
-### @uasge SRC_C_PCLMUL(File Flags...)
+### @usage: SRC_C_PCLMUL(File Flags...)
 ###
 ### Compile a single C/C++ file with PCLMUL and additional Flags
 macro SRC_C_PCLMUL(FILE, FLAGS...) {
@@ -3699,7 +3699,7 @@ macro _SRC_STRICT_C_CPP("C", MACRO_NAME, FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific
-### @uasge SRC_C_PIC(File Flags...)
+### @usage: SRC_C_PIC(File Flags...)
 ###
 ### Compile a single C/C++ file with -fPIC and additional Flags
 macro SRC_C_PIC(FILE, FLAGS...) {
@@ -3707,7 +3707,7 @@ macro SRC_C_PIC(FILE, FLAGS...) {
 }
 
 # tag:src-specific tag:cpp-specific
-### @uasge SRC_C_NO_LTO(File Flags...)
+### @usage: SRC_C_NO_LTO(File Flags...)
 ###
 ### Compile a single C/C++ file with link-time-optimization disabling and additional Flags
 macro SRC_C_NO_LTO(FILE, FLAGS...) {
@@ -3912,7 +3912,7 @@ macro UPDATE_VCS_JAVA_INFO_NODEP(Jar) {
 
 VCS_INFO_FILE=
 
-### @usage VCS_INFO_FILE([FILE out_file])
+### @usage: VCS_INFO_FILE([FILE out_file])
 ###
 ### Enable saving vcs info as a json-file into PACKAGE
 ###
@@ -3951,7 +3951,7 @@ macro DECIMAL_MD5_LOWER_32_BITS(File, FUNCNAME="", Opts...) {
 }
 
 # tag:internal
-### @usage $CFG_VARS # internal
+### @usage: CFG_VARS # internal
 ###
 ### Mark commands that embed Configuration variables into files
 macro CFG_VARS() {
@@ -4379,7 +4379,7 @@ macro _TARGET_SOURCES_FOR_HEADERS(Args...) {
 }
 
 # tag:internal
-### @usage: _TARGET_SOURCES_FOR_HEADERS_IMPL([GENERATE] Args...) # internal
+### @usage: _SET_ENV_FOR_CUSTOM_COMMAND_IMPL([GENERATE] Args...) # internal
 ### Generate prefix " ${CMAKE_COMMAND} -E env " before $Args when GENERATE is specified in the list of actual arguments
 macro _SET_ENV_FOR_CUSTOM_COMMAND_IMPL(GENERATE?" ${CMAKE_COMMAND} -E env ":"", Args...) {
     .SEM=$GENERATE ${pre= :Args}
@@ -4611,7 +4611,7 @@ macro FROM_SANDBOX(Id, OUT{output}[], OUT_NOAUTO{output}[], OUTPUT_INCLUDES[], I
     ADD_CHECK(check.resource $Id)
 }
 
-### @usage LARGE_FILES([AUTOUPDATED]  Files...)
+### @usage: LARGE_FILES([AUTOUPDATED]  Files...)
 ###
 ### Use large file ether from working copy or from remote storage via placeholder <File>.external
 ### If <File> is present locally (and not a symlink!) it will be copied to build directory.
@@ -4814,7 +4814,7 @@ otherwise {
 PACK_TGZ=${cwd:ARCADIA_BUILD_ROOT} tar -czf ${rootrel:OUTPUT} ${rootrel:INPUT} ${hide;kv:"p AR"} ${hide;kv:"pc light-red"}
 
 #  tag:internal
-### @usage TOUCH(Outputs...) # internal
+### @usage: TOUCH(Outputs...) # internal
 ### Just introduce outputs
 macro TOUCH(Outputs...) {
     .CMD=$YMAKE_PYTHON ${input:"build/scripts/touch.py"} ${output:Outputs}
@@ -5132,7 +5132,7 @@ module PREBUILT_PROGRAM: _LINK_UNIT {
     }
 }
 
-### @usage COLLECT_JINJA_TEMPLATES(varname path)
+### @usage: COLLECT_JINJA_TEMPLATES(varname path)
 ###
 ### This macro collects all jinja and yaml files in the directory specified by second argument and
 ### stores result in the variable with mane specified by first parameter.
@@ -5809,7 +5809,7 @@ macro _ADD_CPP_LINTER_CHECK(Args...) {
 }
 
 # tag:test
-### @usage STYLE_CPP([CONFIG_TYPE config_type])
+### @usage: STYLE_CPP([CONFIG_TYPE config_type])
 ###
 ### Run 'ya tool clang-format' test on all cpp sources and headers of the current module
 macro STYLE_CPP(CONFIG_TYPE="") {
diff --git a/contrib/libs/cxxsupp/libsan/generate_symbolizer.py b/contrib/libs/cxxsupp/libsan/generate_symbolizer.py
deleted file mode 100644
index 5c2588bc431..00000000000
--- a/contrib/libs/cxxsupp/libsan/generate_symbolizer.py
+++ /dev/null
@@ -1,12 +0,0 @@
-import os
-import sys
-
-
-def main():
-    print('const char* ya_get_symbolizer_gen() {')
-    print('    return "{}";'.format(os.path.join(os.path.dirname(sys.argv[1]), 'llvm-symbolizer')))
-    print('}')
-
-
-if __name__ == '__main__':
-    main()
diff --git a/contrib/libs/cxxsupp/libsan/inject.c b/contrib/libs/cxxsupp/libsan/inject.c
deleted file mode 100644
index 7790db5f36a..00000000000
--- a/contrib/libs/cxxsupp/libsan/inject.c
+++ /dev/null
@@ -1,8 +0,0 @@
-#include <unistd.h>
-
-extern const char* ya_get_symbolizer_gen();
-
-const char* ya_get_symbolizer() {
-    const char* path = ya_get_symbolizer_gen();
-    return access(path, X_OK) ? NULL : path;
-}
diff --git a/contrib/libs/cxxsupp/libsan/ya.make b/contrib/libs/cxxsupp/libsan/ya.make
deleted file mode 100644
index e770d0ac3df..00000000000
--- a/contrib/libs/cxxsupp/libsan/ya.make
+++ /dev/null
@@ -1,28 +0,0 @@
-LIBRARY()
-
-WITHOUT_LICENSE_TEXTS()
-
-VERSION(Service-proxy-version)
-
-LICENSE(YandexOpen)
-
-NO_PLATFORM()
-
-NO_SANITIZE()
-
-NO_SANITIZE_COVERAGE()
-
-SUBSCRIBER(somov)
-
-RUN_PYTHON3(
-    generate_symbolizer.py ${CXX_COMPILER}
-    STDOUT symbolizer.c
-)
-
-CFLAGS(-fPIC)
-
-SRCS(
-    GLOBAL inject.c
-)
-
-END()
diff --git a/yql/essentials/tests/common/test_framework/test_utils.py b/yql/essentials/tests/common/test_framework/test_utils.py
index 4ec25df5d51..450b5d31398 100644
--- a/yql/essentials/tests/common/test_framework/test_utils.py
+++ b/yql/essentials/tests/common/test_framework/test_utils.py
@@ -56,13 +56,16 @@ def recursive_glob(root, begin_template=None, end_template=None):
             yield os.path.relpath(path, root)
 
 
-def pytest_generate_tests_by_template(template, metafunc):
+def pytest_generate_tests_by_template(template, metafunc, data_path=None):
+    if data_path is None:
+        data_path = DATA_PATH
+
     argvalues = []
 
-    suites = [name for name in os.listdir(DATA_PATH) if os.path.isdir(os.path.join(DATA_PATH, name))]
+    suites = [name for name in os.listdir(data_path) if os.path.isdir(os.path.join(data_path, name))]
     for suite in suites:
         for case in sorted([sql_query_path[:-len(template)]
-                            for sql_query_path in recursive_glob(os.path.join(DATA_PATH, suite), end_template=template)]):
+                            for sql_query_path in recursive_glob(os.path.join(data_path, suite), end_template=template)]):
             argvalues.append((suite, case))
 
     metafunc.parametrize(['suite', 'case'], argvalues)
diff --git a/yql/essentials/tests/common/test_framework/yql_utils.py b/yql/essentials/tests/common/test_framework/yql_utils.py
index a5042129678..dd0be012d51 100644
--- a/yql/essentials/tests/common/test_framework/yql_utils.py
+++ b/yql/essentials/tests/common/test_framework/yql_utils.py
@@ -288,9 +288,11 @@ def normalize_yson(y):
         return [normalize_yson(i) for i in y]
     if isinstance(y, dict):
         return {normalize_yson(k): normalize_yson(v) for k, v in six.iteritems(y)}
-    if not isinstance(y, bytes):
-        return str(y).encode('ascii')
-    return y
+    if isinstance(y, bytes):
+        return y
+    if isinstance(y, six.text_type):
+        return y.encode('utf-8')
+    return str(y).encode('ascii')
 
 
 volatile_attrs = {'DataSize', 'ModifyTime', 'Id', 'Revision'}
diff --git a/yt/cpp/mapreduce/raw_client/rpc_parameters_serialization.cpp b/yt/cpp/mapreduce/raw_client/rpc_parameters_serialization.cpp
index 1ac8257ad44..65bf9e29a5b 100644
--- a/yt/cpp/mapreduce/raw_client/rpc_parameters_serialization.cpp
+++ b/yt/cpp/mapreduce/raw_client/rpc_parameters_serialization.cpp
@@ -190,7 +190,7 @@ TNode SerializeParamsForMultisetAttributes(
     const TTransactionId& transactionId,
     const TString& pathPrefix,
     const TYPath& path,
-    [[maybe_unused]] const TMultisetAttributesOptions& options)
+    const TMultisetAttributesOptions& options)
 {
     TNode result;
     SetTransactionIdParam(&result, transactionId);
diff --git a/yt/python/yt/yson/__init__.py b/yt/python/yt/yson/__init__.py
index ddaec8dd4c0..7ca6f7960be 100644
--- a/yt/python/yt/yson/__init__.py
+++ b/yt/python/yt/yson/__init__.py
@@ -32,7 +32,6 @@ Examples:
 """
 
 from __future__ import print_function
-import os
 
 from . import writer  # noqa
 from . import parser  # noqa
@@ -40,6 +39,7 @@ from . import yson_types  # noqa
 
 TYPE = None
 HAS_PARQUET = False
+HAS_ORC = False
 
 try:
     from yt_yson_bindings import load, loads, dump, dumps # noqa
@@ -54,14 +54,16 @@ except ImportError as error:
         print("Warning! Failed to import YSON bindings: " + message, file=_sys.stderr)
 
 try:
-    from yt_yson_bindings import upload_parquet, dump_parquet, dump_orc, upload_orc, async_dump_parquet, async_dump_orc # noqa
+    from yt_yson_bindings import upload_parquet, dump_parquet, async_dump_parquet # noqa
     HAS_PARQUET = True
-except ImportError as error:
-    message = str(error)
-    if "No module named" not in message:
-        import sys as _sys
-        if os.environ.get("YT_LOG_LEVEL", "").lower() == "debug":
-            print("Warning! Failed to import dump_parquet binding: " + message, file=_sys.stderr)
+except Exception:
+    pass
+
+try:
+    from yt_yson_bindings import dump_orc, upload_orc, async_dump_orc # noqa
+    HAS_ORC = True
+except Exception:
+    pass
 
 if TYPE is None:
     from .parser import load, loads  # noqa
diff --git a/yt/yql/providers/yt/codec/yt_arrow_converter.cpp b/yt/yql/providers/yt/codec/yt_arrow_converter.cpp
index f9cd95019a3..11b512ce2b5 100644
--- a/yt/yql/providers/yt/codec/yt_arrow_converter.cpp
+++ b/yt/yql/providers/yt/codec/yt_arrow_converter.cpp
@@ -686,6 +686,7 @@ std::unique_ptr<IYtColumnConverter> MakeYtColumnConverter(TType* type, const NUd
         case NUdf::EDataSlot::Yson: // Yson there is top-level optional
             // YT type for Yson, Json, String is arrow::Type::BINARY, but yql type is arrow::Type::String
             return std::make_unique<TTopLevelSimpleCastConverter<arrow::Type::BINARY>>(std::move(settings));
+        case NUdf::EDataSlot::Utf8:
         case NUdf::EDataSlot::Double:
         case NUdf::EDataSlot::Int8:
         case NUdf::EDataSlot::Uint8:
diff --git a/yt/yql/providers/yt/provider/yql_yt_datasource_constraints.cpp b/yt/yql/providers/yt/provider/yql_yt_datasource_constraints.cpp
index 3c5d2f4554d..8eada7e7066 100644
--- a/yt/yql/providers/yt/provider/yql_yt_datasource_constraints.cpp
+++ b/yt/yql/providers/yt/provider/yql_yt_datasource_constraints.cpp
@@ -99,19 +99,19 @@ public:
 
         if (const auto sort = path.Table().Ref().GetConstraint<TSortedConstraintNode>()) {
             if (const auto filtered = sort->FilterFields(ctx, filter)) {
-                path.Ptr()->AddConstraint(filtered);
+                path.Ptr()->AddConstraint(filtered->GetSimplifiedForType(*path.Ref().GetTypeAnn(), ctx));
             }
         }
 
         if (const auto uniq = path.Table().Ref().GetConstraint<TUniqueConstraintNode>()) {
             if (const auto filtered = uniq->FilterFields(ctx, filter)) {
-                path.Ptr()->AddConstraint(filtered);
+                path.Ptr()->AddConstraint(filtered->GetSimplifiedForType(*path.Ref().GetTypeAnn(), ctx));
             }
         }
 
         if (const auto dist = path.Table().Ref().GetConstraint<TDistinctConstraintNode>()) {
             if (const auto filtered = dist->FilterFields(ctx, filter)) {
-                path.Ptr()->AddConstraint(filtered);
+                path.Ptr()->AddConstraint(filtered->GetSimplifiedForType(*path.Ref().GetTypeAnn(), ctx));
             }
         }
 
diff --git a/yt/yt/client/api/delegating_client.h b/yt/yt/client/api/delegating_client.h
index 62048663c33..ec546404b5d 100644
--- a/yt/yt/client/api/delegating_client.h
+++ b/yt/yt/client/api/delegating_client.h
@@ -846,19 +846,19 @@ public:
         (pipelinePath, viewPath, options))
 
     // Distributed client
-    DELEGATE_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (
+    DELEGATE_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options),
         (path, options))
 
     DELEGATE_METHOD(TFuture<void>, FinishDistributedWriteSession, (
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options),
-        (std::move(session), options))
+        (sessionWithResults, options))
 
-    DELEGATE_METHOD(TFuture<ITableWriterPtr>, CreateFragmentTableWriter, (
-        const TFragmentWriteCookiePtr& cookie,
-        const TFragmentTableWriterOptions& options),
+    DELEGATE_METHOD(TFuture<ITableFragmentWriterPtr>, CreateTableFragmentWriter, (
+        const TSignedWriteFragmentCookiePtr& cookie,
+        const TTableFragmentWriterOptions& options),
         (cookie, options))
 
     // Shuffle Service
diff --git a/yt/yt/client/api/delegating_transaction.cpp b/yt/yt/client/api/delegating_transaction.cpp
index 8a8ddf96dd9..2e12a18cfb4 100644
--- a/yt/yt/client/api/delegating_transaction.cpp
+++ b/yt/yt/client/api/delegating_transaction.cpp
@@ -312,15 +312,15 @@ DELEGATE_METHOD(TFuture<TPushQueueProducerResult>, PushQueueProducer, (
     const TPushQueueProducerOptions& options),
     (producerPath, queuePath, sessionId, epoch, nameTable, serializedRows, options))
 
-DELEGATE_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (
+DELEGATE_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (
     const NYPath::TRichYPath& path,
     const TDistributedWriteSessionStartOptions& options),
     (path, options))
 
 DELEGATE_METHOD(TFuture<void>, FinishDistributedWriteSession, (
-    TDistributedWriteSessionPtr session,
+    const TDistributedWriteSessionWithResults& sessionWithResults,
     const TDistributedWriteSessionFinishOptions& options),
-    (std::move(session), options))
+    (sessionWithResults, options))
 
 #undef DELEGATE_METHOD
 
diff --git a/yt/yt/client/api/delegating_transaction.h b/yt/yt/client/api/delegating_transaction.h
index e0263fc7992..5823e143fc9 100644
--- a/yt/yt/client/api/delegating_transaction.h
+++ b/yt/yt/client/api/delegating_transaction.h
@@ -257,12 +257,12 @@ public:
         const TPushQueueProducerOptions& options) override;
 
     // Distributed table client
-    TFuture<TDistributedWriteSessionPtr> StartDistributedWriteSession(
+    TFuture<TDistributedWriteSessionWithCookies> StartDistributedWriteSession(
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options = {}) override;
 
     TFuture<void> FinishDistributedWriteSession(
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options = {}) override;
 
 protected:
diff --git a/yt/yt/client/api/distributed_table_client.h b/yt/yt/client/api/distributed_table_client.h
index 7417c5b620d..3cb3640cb0a 100644
--- a/yt/yt/client/api/distributed_table_client.h
+++ b/yt/yt/client/api/distributed_table_client.h
@@ -2,24 +2,44 @@
 
 #include "table_client.h"
 
-#include <yt/yt/client/table_client/config.h>
+#include <yt/yt/client/signature/public.h>
 
-#include <library/cpp/yt/memory/non_null_ptr.h>
+#include <yt/yt/client/table_client/config.h>
 
 namespace NYT::NApi {
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct TDistributedWriteSessionWithCookies
+{
+    // TDistributedWriteSession.
+    TSignedDistributedWriteSessionPtr Session;
+    // std::vector<TWriteFragmentCookie>.
+    std::vector<TSignedWriteFragmentCookiePtr> Cookies;
+};
+
+struct TDistributedWriteSessionWithResults
+{
+    // TDistributedWriteSession.
+    TSignedDistributedWriteSessionPtr Session;
+    // std::vector<TWriteFragmentResult>.
+    std::vector<TSignedWriteFragmentResultPtr> Results;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
 struct TDistributedWriteSessionStartOptions
     : public TTransactionalOptions
-{ };
+{
+    int CookieCount = 0;
+};
 
 struct TDistributedWriteSessionFinishOptions
 {
     int MaxChildrenPerAttachRequest = 10'000;
 };
 
-struct TFragmentTableWriterOptions
+struct TTableFragmentWriterOptions
     : public TTableWriterOptions
 { };
 
@@ -27,23 +47,15 @@ struct TFragmentTableWriterOptions
 
 struct IDistributedTableClientBase
 {
-public:
     virtual ~IDistributedTableClientBase() = default;
 
-    virtual TFuture<TDistributedWriteSessionPtr> StartDistributedWriteSession(
+    virtual TFuture<TDistributedWriteSessionWithCookies> StartDistributedWriteSession(
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options = {}) = 0;
 
     virtual TFuture<void> FinishDistributedWriteSession(
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options = {}) = 0;
-
-    // Helper used to implement FinishDistributedWriteSession efficiently
-    // without compromising privacy of session fields.
-    // defined in yt/yt/client/api/distributed_table_session.cpp.
-    void* GetOpaqueDistributedWriteResults(Y_LIFETIME_BOUND const TDistributedWriteSessionPtr& session);
-    // Used in chunk writer for results recording
-    void RecordOpaqueWriteResult(const TFragmentWriteCookiePtr& cookie, void* opaqueWriteResult);
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -52,11 +64,18 @@ struct IDistributedTableClient
 {
     virtual ~IDistributedTableClient() = default;
 
-    virtual TFuture<ITableWriterPtr> CreateFragmentTableWriter(
-        const TFragmentWriteCookiePtr& cookie,
-        const TFragmentTableWriterOptions& options = {}) = 0;
+    virtual TFuture<ITableFragmentWriterPtr> CreateTableFragmentWriter(
+        const TSignedWriteFragmentCookiePtr& cookie,
+        const TTableFragmentWriterOptions& options = {}) = 0;
 };
 
 ////////////////////////////////////////////////////////////////////////////////
 
+// Defined in distributed_table_session.cpp.
+TFuture<void> PingDistributedWriteSession(
+    const TSignedDistributedWriteSessionPtr& session,
+    const IClientPtr& client);
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NApi
diff --git a/yt/yt/client/api/distributed_table_session.cpp b/yt/yt/client/api/distributed_table_session.cpp
index d67998b9d21..555c5f68129 100644
--- a/yt/yt/client/api/distributed_table_session.cpp
+++ b/yt/yt/client/api/distributed_table_session.cpp
@@ -1,8 +1,11 @@
 #include "distributed_table_session.h"
 
 #include "client.h"
+#include "distributed_table_client.h"
 #include "transaction.h"
 
+#include <yt/yt/client/signature/signature.h>
+
 namespace NYT::NApi {
 
 using namespace NYPath;
@@ -15,34 +18,6 @@ using namespace NChunkClient;
 
 ////////////////////////////////////////////////////////////////////////////////
 
-TTableWriterPatchInfo::TTableWriterPatchInfo(
-    TRichYPath richPath,
-    TObjectId objectId,
-    TCellTag externalCellTag,
-    TMasterTableSchemaId chunkSchemaId,
-    TTableSchemaPtr chunkSchema,
-    std::optional<TLegacyOwningKey> writerLastKey,
-    int maxHeavyColumns,
-    TTimestamp timestamp,
-    const IAttributeDictionary& tableAttributes)
-    : TTableWriterPatchInfo()
-{
-    ObjectId = objectId;
-    RichPath = std::move(richPath);
-
-    ChunkSchemaId = chunkSchemaId;
-    ChunkSchema = std::move(chunkSchema);
-
-    WriterLastKey = writerLastKey;
-    MaxHeavyColumns = maxHeavyColumns;
-
-    TableAttributes = tableAttributes.ToMap();
-
-    ExternalCellTag = externalCellTag;
-
-    Timestamp = timestamp;
-}
-
 void TTableWriterPatchInfo::Register(TRegistrar registrar)
 {
     registrar.Parameter("table_id", &TThis::ObjectId);
@@ -63,8 +38,10 @@ void TTableWriterPatchInfo::Register(TRegistrar registrar)
 
 ////////////////////////////////////////////////////////////////////////////////
 
-void TFragmentWriteResult::Register(TRegistrar registrar)
+void TWriteFragmentResult::Register(TRegistrar registrar)
 {
+    registrar.Parameter("session_id", &TThis::SessionId);
+    registrar.Parameter("cookie_id", &TThis::CookieId);
     registrar.Parameter("min_boundary_key", &TThis::MinBoundaryKey);
     registrar.Parameter("max_boundary_key", &TThis::MaxBoundaryKey);
     registrar.Parameter("chunk_list_id", &TThis::ChunkListId);
@@ -72,126 +49,89 @@ void TFragmentWriteResult::Register(TRegistrar registrar)
 
 ////////////////////////////////////////////////////////////////////////////////
 
-const TTableWriterPatchInfo& TFragmentWriteCookie::GetPatchInfo() const
+void TWriteFragmentCookie::Register(TRegistrar registrar)
 {
-    return PatchInfo_;
-}
+    registrar.Parameter("session_id", &TThis::SessionId);
+    registrar.Parameter("cookie_id", &TThis::CookieId);
 
-TTransactionId TFragmentWriteCookie::GetMainTransactionId() const
-{
-    return MainTxId_;
-}
+    registrar.Parameter("transaction_id", &TThis::MainTransactionId);
 
-TTransactionId TFragmentWriteCookie::GetUploadTransactionId() const
-{
-    return UploadTxId_;
-}
-
-void TFragmentWriteCookie::Register(TRegistrar registrar)
-{
-    registrar.Parameter("session_id", &TThis::Id_);
-
-    registrar.Parameter("tx_id", &TThis::MainTxId_);
-    registrar.Parameter("upload_tx_id", &TThis::UploadTxId_);
-
-    registrar.Parameter("patch_info", &TThis::PatchInfo_);
-
-    registrar.Parameter("writer_results", &TThis::WriteResults_);
+    registrar.Parameter("patch_info", &TThis::PatchInfo);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
 TDistributedWriteSession::TDistributedWriteSession(
-    TTransactionId mainTxId,
-    TTransactionId uploadTxId,
+    TTransactionId mainTransactionId,
+    TTransactionId uploadTransactionId,
     TChunkListId rootChunkListId,
-    TTableWriterPatchInfo patchInfo)
+    NYPath::TRichYPath richPath,
+    NObjectClient::TObjectId objectId,
+    NObjectClient::TCellTag externalCellTag,
+    NTableClient::TMasterTableSchemaId chunkSchemaId,
+    NTableClient::TTableSchemaPtr chunkSchema,
+    std::optional<NTableClient::TLegacyOwningKey> writerLastKey,
+    int maxHeavyColumns,
+    NTransactionClient::TTimestamp timestamp,
+    const NYTree::IAttributeDictionary& tableAttributes)
     : TDistributedWriteSession()
 {
-    Id_ = TDistributedWriteSessionId(TGuid::Create());
-    MainTxId_ = mainTxId;
-    UploadTxId_ = uploadTxId;
+    MainTransactionId = mainTransactionId;
+    UploadTransactionId = uploadTransactionId;
 
-    RootChunkListId_ = rootChunkListId;
+    RootChunkListId = rootChunkListId;
 
-    PatchInfo_ = std::move(patchInfo);
+    PatchInfo.ObjectId = objectId;
+    PatchInfo.RichPath = std::move(richPath);
 
-    // Signatures?
-}
+    PatchInfo.ChunkSchemaId = chunkSchemaId;
+    PatchInfo.ChunkSchema = std::move(chunkSchema);
 
-TTransactionId TDistributedWriteSession::GetMainTransactionId() const
-{
-    return MainTxId_;
-}
+    PatchInfo.WriterLastKey = std::move(writerLastKey);
+    PatchInfo.MaxHeavyColumns = maxHeavyColumns;
 
-TTransactionId TDistributedWriteSession::GetUploadTransactionId() const
-{
-    return UploadTxId_;
-}
+    PatchInfo.TableAttributes = tableAttributes.ToMap();
 
-const TTableWriterPatchInfo& TDistributedWriteSession::GetPatchInfo() const Y_LIFETIME_BOUND
-{
-    return PatchInfo_;
-}
+    PatchInfo.ExternalCellTag = externalCellTag;
 
-TChunkListId TDistributedWriteSession::GetRootChunkListId() const
-{
-    return RootChunkListId_;
+    PatchInfo.Timestamp = timestamp;
 }
 
-TFragmentWriteCookiePtr TDistributedWriteSession::GiveCookie()
+TWriteFragmentCookie TDistributedWriteSession::CookieFromThis() const
 {
-    auto cookie = New<TFragmentWriteCookie>();
-    cookie->Id_ = Id_;
-    cookie->MainTxId_ = MainTxId_;
-    cookie->UploadTxId_ = UploadTxId_;
-    cookie->PatchInfo_ = PatchInfo_;
+    auto cookie = TWriteFragmentCookie{};
+    cookie.CookieId = TGuid::Create();
+    cookie.SessionId = RootChunkListId;
+    cookie.MainTransactionId = MainTransactionId;
+    cookie.PatchInfo = PatchInfo;
 
     return cookie;
 }
 
-void TDistributedWriteSession::TakeCookie(TFragmentWriteCookiePtr cookie)
-{
-    // Verify cookie signature?
-    WriteResults_.reserve(std::ssize(WriteResults_) + std::ssize(cookie->WriteResults_));
-    for (const auto& writeResult : cookie->WriteResults_) {
-        WriteResults_.push_back(writeResult);
-    }
-}
-
-TFuture<void> TDistributedWriteSession::Ping(IClientPtr client)
-{
-    // NB(arkady-e1ppa): AutoAbort = false by default.
-    auto mainTx = client->AttachTransaction(MainTxId_);
-
-    return mainTx->Ping();
-}
-
 void TDistributedWriteSession::Register(TRegistrar registrar)
 {
-    registrar.Parameter("session_id", &TThis::Id_);
-    registrar.Parameter("tx_id", &TThis::MainTxId_);
-    registrar.Parameter("upload_tx_id", &TThis::UploadTxId_);
+    registrar.Parameter("main_transaction_id", &TThis::MainTransactionId);
+    registrar.Parameter("upload_transaction_id", &TThis::UploadTransactionId);
 
-    registrar.Parameter("root_chunk_list_id", &TThis::RootChunkListId_);
+    registrar.Parameter("root_chunk_list_id", &TThis::RootChunkListId);
 
-    registrar.Parameter("patch_info", &TThis::PatchInfo_);
-
-    registrar.Parameter("write_results", &TThis::WriteResults_);
+    registrar.Parameter("patch_info", &TThis::PatchInfo);
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
-void* IDistributedTableClientBase::GetOpaqueDistributedWriteResults(Y_LIFETIME_BOUND const TDistributedWriteSessionPtr& session)
+TFuture<void> PingDistributedWriteSession(
+    const TSignedDistributedWriteSessionPtr& session,
+    const IClientPtr& client)
 {
-    return static_cast<void*>(&session->WriteResults_);
-}
+    auto concreteSession = ConvertTo<TDistributedWriteSession>(session.Underlying()->Payload());
 
-void IDistributedTableClientBase::RecordOpaqueWriteResult(const TFragmentWriteCookiePtr& cookie, void* opaqueWriteResult)
-{
-    auto* concrete = static_cast<TFragmentWriteResult*>(opaqueWriteResult);
-    YT_ASSERT(concrete);
-    cookie->WriteResults_.push_back(*concrete);
+    // NB(arkady-e1ppa): AutoAbort = false by default.
+    auto mainTx = client->AttachTransaction(concreteSession.MainTransactionId);
+
+    return mainTx->Ping();
 }
 
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NApi
diff --git a/yt/yt/client/api/distributed_table_session.h b/yt/yt/client/api/distributed_table_session.h
index c8c074c84c6..136ffcfcd4e 100644
--- a/yt/yt/client/api/distributed_table_session.h
+++ b/yt/yt/client/api/distributed_table_session.h
@@ -16,10 +16,6 @@ namespace NYT::NApi {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-YT_DEFINE_STRONG_TYPEDEF(TDistributedWriteSessionId, TGuid);
-
-////////////////////////////////////////////////////////////////////////////////
-
 // Used by distributed table writer to patch
 // its config based on table data.
 // NB(arkady-e1ppa): TableUploadOptions are
@@ -28,7 +24,6 @@ YT_DEFINE_STRONG_TYPEDEF(TDistributedWriteSessionId, TGuid);
 struct TTableWriterPatchInfo
     : public NYTree::TYsonStructLite
 {
-public:
     NObjectClient::TObjectId ObjectId;
 
     NYPath::TRichYPath RichPath;
@@ -48,17 +43,6 @@ public:
 
     NTransactionClient::TTimestamp Timestamp;
 
-    TTableWriterPatchInfo(
-        NYPath::TRichYPath richPath,
-        NObjectClient::TObjectId objectId,
-        NObjectClient::TCellTag externalCellTag,
-        NTableClient::TMasterTableSchemaId chunkSchemaId,
-        NTableClient::TTableSchemaPtr chunkSchema,
-        std::optional<NTableClient::TLegacyOwningKey> writerLastKey,
-        int maxHeavyColumns,
-        NTransactionClient::TTimestamp timestamp,
-        const NYTree::IAttributeDictionary& tableAttributes);
-
     REGISTER_YSON_STRUCT_LITE(TTableWriterPatchInfo)
 
     static void Register(TRegistrar registrar);
@@ -66,93 +50,73 @@ public:
 
 ////////////////////////////////////////////////////////////////////////////////
 
-struct TFragmentWriteResult
+struct TWriteFragmentResult
     : public NYTree::TYsonStructLite
 {
+    TGuid SessionId;
+    TGuid CookieId;
+
     NTableClient::TLegacyOwningKey MinBoundaryKey;
     NTableClient::TLegacyOwningKey MaxBoundaryKey;
     NChunkClient::TChunkListId ChunkListId;
 
-    REGISTER_YSON_STRUCT_LITE(TFragmentWriteResult)
+    REGISTER_YSON_STRUCT_LITE(TWriteFragmentResult)
 
     static void Register(TRegistrar registrar);
 };
 
 ////////////////////////////////////////////////////////////////////////////////
 
-class TFragmentWriteCookie
-    : public NYTree::TYsonStruct
+struct TWriteFragmentCookie
+    : public NYTree::TYsonStructLite
 {
-public:
-    const TTableWriterPatchInfo& GetPatchInfo() const;
-
-    NCypressClient::TTransactionId GetMainTransactionId() const;
-    NCypressClient::TTransactionId GetUploadTransactionId() const;
+    TGuid SessionId;
+    TGuid CookieId;
 
-private:
-    TDistributedWriteSessionId Id_;
+    NCypressClient::TTransactionId MainTransactionId;
 
-    NCypressClient::TTransactionId MainTxId_;
-    NCypressClient::TTransactionId UploadTxId_;
+    TTableWriterPatchInfo PatchInfo;
 
-    TTableWriterPatchInfo PatchInfo_;
-
-    std::vector<TFragmentWriteResult> WriteResults_;
-
-    REGISTER_YSON_STRUCT(TFragmentWriteCookie);
+    REGISTER_YSON_STRUCT_LITE(TWriteFragmentCookie);
 
     static void Register(TRegistrar registrar);
-
-    friend class TDistributedWriteSession;
-    friend struct IDistributedTableClientBase;
 };
 
-DEFINE_REFCOUNTED_TYPE(TFragmentWriteCookie);
-
 ////////////////////////////////////////////////////////////////////////////////
 
-class TDistributedWriteSession
-    : public NYTree::TYsonStruct
+struct TDistributedWriteSession
+    : public NYTree::TYsonStructLite
 {
-public:
-    TDistributedWriteSession(
-        NCypressClient::TTransactionId mainTxId,
-        NCypressClient::TTransactionId uploadTxId,
-        NChunkClient::TChunkListId rootChunkListId,
-        TTableWriterPatchInfo patchInfo);
-
-    NCypressClient::TTransactionId GetMainTransactionId() const;
-    NCypressClient::TTransactionId GetUploadTransactionId() const;
-    const TTableWriterPatchInfo& GetPatchInfo() const Y_LIFETIME_BOUND;
-    NChunkClient::TChunkListId GetRootChunkListId() const;
-
-    TFragmentWriteCookiePtr GiveCookie();
-    void TakeCookie(TFragmentWriteCookiePtr cookie);
-
-    TFuture<void> Ping(IClientPtr client);
 
-    REGISTER_YSON_STRUCT(TDistributedWriteSession);
+    // RootChunkListId is used for indentification.
+    NChunkClient::TChunkListId RootChunkListId;
 
-    static void Register(TRegistrar registrar);
-
-private:
-    TDistributedWriteSessionId Id_;
+    NCypressClient::TTransactionId MainTransactionId;
+    NCypressClient::TTransactionId UploadTransactionId;
 
-    NCypressClient::TTransactionId MainTxId_;
-    NCypressClient::TTransactionId UploadTxId_;
+    TTableWriterPatchInfo PatchInfo;
 
-    NChunkClient::TChunkListId RootChunkListId_;
+    TDistributedWriteSession(
+        NCypressClient::TTransactionId mainTransactionId,
+        NCypressClient::TTransactionId uploadTransactionId,
+        NChunkClient::TChunkListId rootChunkListId,
+        NYPath::TRichYPath richPath,
+        NObjectClient::TObjectId objectId,
+        NObjectClient::TCellTag externalCellTag,
+        NTableClient::TMasterTableSchemaId chunkSchemaId,
+        NTableClient::TTableSchemaPtr chunkSchema,
+        std::optional<NTableClient::TLegacyOwningKey> writerLastKey,
+        int maxHeavyColumns,
+        NTransactionClient::TTimestamp timestamp,
+        const NYTree::IAttributeDictionary& tableAttributes);
 
-    TTableWriterPatchInfo PatchInfo_;
+    TWriteFragmentCookie CookieFromThis() const;
 
-    // This is used to commit changes when session is over.
-    std::vector<TFragmentWriteResult> WriteResults_;
+    REGISTER_YSON_STRUCT_LITE(TDistributedWriteSession);
 
-    friend struct IDistributedTableClientBase;
+    static void Register(TRegistrar registrar);
 };
 
-DEFINE_REFCOUNTED_TYPE(TDistributedWriteSession);
-
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NApi
diff --git a/yt/yt/client/api/public.h b/yt/yt/client/api/public.h
index 0e92ead601c..a8293e640ed 100644
--- a/yt/yt/client/api/public.h
+++ b/yt/yt/client/api/public.h
@@ -146,6 +146,8 @@ DECLARE_REFCOUNTED_STRUCT(IRowBatchWriter)
 DECLARE_REFCOUNTED_STRUCT(ITableReader)
 DECLARE_REFCOUNTED_STRUCT(ITableWriter)
 
+DECLARE_REFCOUNTED_STRUCT(ITableFragmentWriter);
+
 DECLARE_REFCOUNTED_STRUCT(IFileReader)
 DECLARE_REFCOUNTED_STRUCT(IFileWriter)
 
@@ -187,10 +189,6 @@ DECLARE_REFCOUNTED_STRUCT(TBackupManifest)
 
 DECLARE_REFCOUNTED_STRUCT(TListOperationsAccessFilter)
 
-DECLARE_REFCOUNTED_CLASS(TDistributedWriteSession)
-DECLARE_REFCOUNTED_CLASS(TFragmentWriteCookie)
-struct IDistributedTableClientBase;
-
 DECLARE_REFCOUNTED_STRUCT(TShuffleHandle)
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -240,4 +238,11 @@ using TMaintenanceCountsPerTarget = TCompactFlatMap<std::string, TMaintenanceCou
 
 ////////////////////////////////////////////////////////////////////////////////
 
+using NTableClient::TSignedDistributedWriteSessionPtr;
+using NTableClient::TSignedWriteFragmentCookiePtr;
+using NTableClient::TSignedWriteFragmentResultPtr;
+struct TWriteFragmentCookie;
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NApi
diff --git a/yt/yt/client/api/rpc_proxy/client_base.cpp b/yt/yt/client/api/rpc_proxy/client_base.cpp
index 58b2538df6b..26062e93ee0 100644
--- a/yt/yt/client/api/rpc_proxy/client_base.cpp
+++ b/yt/yt/client/api/rpc_proxy/client_base.cpp
@@ -21,6 +21,8 @@
 
 #include <yt/yt/client/chaos_client/replication_card_serialization.h>
 
+#include <yt/yt/client/signature/signature.h>
+
 #include <yt/yt/client/table_client/name_table.h>
 #include <yt/yt/client/table_client/row_base.h>
 #include <yt/yt/client/table_client/row_buffer.h>
@@ -792,7 +794,7 @@ TFuture<ITableWriterPtr> TClientBase::CreateTableWriter(
 
 ////////////////////////////////////////////////////////////////////////////////
 
-TFuture<TDistributedWriteSessionPtr> TClientBase::StartDistributedWriteSession(
+TFuture<TDistributedWriteSessionWithCookies> TClientBase::StartDistributedWriteSession(
     const NYPath::TRichYPath& path,
     const TDistributedWriteSessionStartOptions& options)
 {
@@ -804,21 +806,28 @@ TFuture<TDistributedWriteSessionPtr> TClientBase::StartDistributedWriteSession(
     FillRequest(req.Get(), path, options);
 
     return req->Invoke()
-        .ApplyUnique(BIND([] (TRsp&& result) -> TDistributedWriteSessionPtr {
-            return ConvertTo<TDistributedWriteSessionPtr>(TYsonString(result->session()));
+        .ApplyUnique(BIND([] (TRsp&& result) -> TDistributedWriteSessionWithCookies {
+            std::vector<TSignedWriteFragmentCookiePtr> cookies;
+            cookies.reserve(result->signed_cookies().size());
+            for (const auto& cookie : result->signed_cookies()) {
+                cookies.push_back(ConvertTo<TSignedWriteFragmentCookiePtr>(TYsonString(cookie)));
+            }
+            return TDistributedWriteSessionWithCookies{
+                .Session = ConvertTo<TSignedDistributedWriteSessionPtr>(TYsonString(result->signed_session())),
+                .Cookies = std::move(cookies),
+            };
         }));
 }
 
 TFuture<void> TClientBase::FinishDistributedWriteSession(
-    TDistributedWriteSessionPtr session,
+    const TDistributedWriteSessionWithResults& sessionWithResults,
     const TDistributedWriteSessionFinishOptions& options)
 {
     auto proxy = CreateApiServiceProxy();
 
     auto req = proxy.FinishDistributedWriteSession();
 
-    FillRequest(req.Get(), std::move(session), options);
-
+    FillRequest(req.Get(), sessionWithResults, options);
     return req->Invoke().AsVoid();
 }
 
diff --git a/yt/yt/client/api/rpc_proxy/client_base.h b/yt/yt/client/api/rpc_proxy/client_base.h
index 2ccdb1adb49..424752d95e1 100644
--- a/yt/yt/client/api/rpc_proxy/client_base.h
+++ b/yt/yt/client/api/rpc_proxy/client_base.h
@@ -185,12 +185,12 @@ public:
         const NApi::TTableWriterOptions& options) override;
 
     // Distributed table client
-    TFuture<TDistributedWriteSessionPtr> StartDistributedWriteSession(
+    TFuture<TDistributedWriteSessionWithCookies> StartDistributedWriteSession(
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options) override;
 
     TFuture<void> FinishDistributedWriteSession(
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options) override;
 };
 
diff --git a/yt/yt/client/api/rpc_proxy/client_impl.cpp b/yt/yt/client/api/rpc_proxy/client_impl.cpp
index 3ab8c239d51..5c90b89d185 100644
--- a/yt/yt/client/api/rpc_proxy/client_impl.cpp
+++ b/yt/yt/client/api/rpc_proxy/client_impl.cpp
@@ -16,6 +16,8 @@
 
 #include <yt/yt/client/scheduler/operation_id_or_alias.h>
 
+#include <yt/yt/client/signature/signature.h>
+
 #include <yt/yt/client/table_client/columnar_statistics.h>
 #include <yt/yt/client/table_client/schema.h>
 #include <yt/yt/client/table_client/unversioned_row.h>
@@ -742,11 +744,12 @@ TFuture<void> TClient::AlterReplicationCard(
     return req->Invoke().As<void>();
 }
 
-TFuture<ITableWriterPtr> TClient::CreateFragmentTableWriter(
-    const TFragmentWriteCookiePtr& cookie,
-    const TFragmentTableWriterOptions& options)
+TFuture<ITableFragmentWriterPtr> TClient::CreateTableFragmentWriter(
+    const TSignedWriteFragmentCookiePtr& cookie,
+    const TTableFragmentWriterOptions& options)
 {
     using TRspPtr = TIntrusivePtr<NRpc::TTypedClientResponse<NProto::TRspWriteTableFragment>>;
+    YT_VERIFY(cookie);
 
     auto proxy = CreateApiServiceProxy();
     auto req = proxy.WriteTableFragment();
@@ -755,22 +758,29 @@ TFuture<ITableWriterPtr> TClient::CreateFragmentTableWriter(
     FillRequest(req.Get(), cookie, options);
 
     auto schema = New<TTableSchema>();
+    auto promise = NewPromise<TSignedWriteFragmentResultPtr>();
+
+    // NB(arkady-e1ppa): Whenever stream is over, rsp future is set
+    // with the value of write result. We create a channel via promise-future
+    // to transfer this write result to the TableWriter adapter. In order to avoid races
+    // when TableWriter is already closed (and so is stream) but the promise with
+    // write result is not yet set, consider writer closed only after said promise is set.
     return CreateRpcClientOutputStream(
         std::move(req),
         BIND ([=] (const TSharedRef& metaRef) {
             NApi::NRpcProxy::NProto::TWriteTableMeta meta;
             if (!TryDeserializeProto(&meta, metaRef)) {
-                THROW_ERROR_EXCEPTION("Failed to deserialize schema for fragment table writer");
+                THROW_ERROR_EXCEPTION("Failed to deserialize schema for table fragment writer");
             }
 
             FromProto(schema.Get(), meta.schema());
         }),
-        BIND([=] (TRspPtr&& rsp) mutable {
-            Deserialize(*cookie, ConvertToNode(TYsonString(rsp->cookie())));
+        BIND([=] (TRspPtr&& rsp)  {
+            promise.Set(ConvertTo<TSignedWriteFragmentResultPtr>(TYsonString(rsp->signed_write_result())));
         }))
-        .ApplyUnique(BIND([=] (IAsyncZeroCopyOutputStreamPtr&& outputStream) {
-            return NRpcProxy::CreateTableWriter(std::move(outputStream), std::move(schema));
-        })).As<ITableWriterPtr>();
+        .ApplyUnique(BIND([=, future = promise.ToFuture()] (IAsyncZeroCopyOutputStreamPtr&& outputStream) {
+            return NRpcProxy::CreateTableFragmentWriter(std::move(outputStream), std::move(schema), std::move(future));
+        }));
 }
 
 TFuture<IQueueRowsetPtr> TClient::PullQueue(
diff --git a/yt/yt/client/api/rpc_proxy/client_impl.h b/yt/yt/client/api/rpc_proxy/client_impl.h
index 4a5cc37d792..a161cfe88dc 100644
--- a/yt/yt/client/api/rpc_proxy/client_impl.h
+++ b/yt/yt/client/api/rpc_proxy/client_impl.h
@@ -130,9 +130,9 @@ public:
         const TAlterReplicationCardOptions& options = {}) override;
 
     // Distributed table client
-    TFuture<ITableWriterPtr> CreateFragmentTableWriter(
-        const TFragmentWriteCookiePtr& cookie,
-        const TFragmentTableWriterOptions& options) override;
+    TFuture<ITableFragmentWriterPtr> CreateTableFragmentWriter(
+        const TSignedWriteFragmentCookiePtr& cookie,
+        const TTableFragmentWriterOptions& options) override;
 
     // Queues.
     TFuture<NQueueClient::IQueueRowsetPtr> PullQueue(
diff --git a/yt/yt/client/api/rpc_proxy/helpers.cpp b/yt/yt/client/api/rpc_proxy/helpers.cpp
index 9c558cfdf2d..5baa8460ee1 100644
--- a/yt/yt/client/api/rpc_proxy/helpers.cpp
+++ b/yt/yt/client/api/rpc_proxy/helpers.cpp
@@ -6,6 +6,8 @@
 
 #include <yt/yt/client/sequoia_client/public.h>
 
+#include <yt/yt/client/signature/signature.h>
+
 #include <yt/yt/client/table_client/columnar_statistics.h>
 #include <yt/yt/client/table_client/column_sort_schema.h>
 #include <yt/yt/client/table_client/logical_type.h>
@@ -1964,6 +1966,7 @@ void FillRequest(
     const TDistributedWriteSessionStartOptions& options)
 {
     ToProto(req->mutable_path(), path);
+    req->set_cookie_count(options.CookieCount);
 
     if (options.TransactionId) {
         ToProto(req->mutable_transactional_options(), options);
@@ -1976,6 +1979,7 @@ void ParseRequest(
     const TReqStartDistributedWriteSession& req)
 {
     *mutablePath = FromProto<NYPath::TRichYPath>(req.path());
+    mutableOptions->CookieCount = req.cookie_count();
     if (req.has_transactional_options()) {
         FromProto(mutableOptions, req.transactional_options());
     }
@@ -1985,19 +1989,31 @@ void ParseRequest(
 
 void FillRequest(
     TReqFinishDistributedWriteSession* req,
-    TDistributedWriteSessionPtr session,
+    const TDistributedWriteSessionWithResults& sessionWithResults,
     const TDistributedWriteSessionFinishOptions& options)
 {
-    req->set_session(ConvertToYsonString(session).ToString());
+    YT_VERIFY(sessionWithResults.Session);
+
+    req->set_signed_session(ConvertToYsonString(sessionWithResults.Session).ToString());
+    for (const auto& writeResult : sessionWithResults.Results) {
+        YT_VERIFY(writeResult);
+        req->add_signed_write_results(ConvertToYsonString(writeResult).ToString());
+    }
     req->set_max_children_per_attach_request(options.MaxChildrenPerAttachRequest);
 }
 
 void ParseRequest(
-    TDistributedWriteSessionPtr* mutableSession,
+    TDistributedWriteSessionWithResults* mutableSessionWithResults,
     TDistributedWriteSessionFinishOptions* mutableOptions,
     const TReqFinishDistributedWriteSession& req)
 {
-    *mutableSession = ConvertTo<TDistributedWriteSessionPtr>(TYsonString(req.session()));
+    mutableSessionWithResults->Results.reserve(req.signed_write_results().size());
+    for (const auto& writeResult : req.signed_write_results()) {
+        mutableSessionWithResults->Results.push_back(ConvertTo<TSignedWriteFragmentResultPtr>(TYsonString(writeResult)));
+    }
+
+    mutableSessionWithResults->Session = ConvertTo<TSignedDistributedWriteSessionPtr>(TYsonString(req.signed_session()));
+
     mutableOptions->MaxChildrenPerAttachRequest = req.max_children_per_attach_request();
 }
 
@@ -2005,10 +2021,10 @@ void ParseRequest(
 
 void FillRequest(
     TReqWriteTableFragment* req,
-    const TFragmentWriteCookiePtr& cookie,
-    const TFragmentTableWriterOptions& options)
+    const TSignedWriteFragmentCookiePtr& cookie,
+    const TTableFragmentWriterOptions& options)
 {
-    req->set_cookie(ConvertToYsonString(cookie).ToString());
+    req->set_signed_cookie(ConvertToYsonString(cookie).ToString());
 
     if (options.Config) {
         req->set_config(ConvertToYsonString(*options.Config).ToString());
@@ -2016,11 +2032,11 @@ void FillRequest(
 }
 
 void ParseRequest(
-    TFragmentWriteCookiePtr* mutableCookie,
-    TFragmentTableWriterOptions* mutableOptions,
+    TSignedWriteFragmentCookiePtr* mutableCookie,
+    TTableFragmentWriterOptions* mutableOptions,
     const TReqWriteTableFragment& req)
 {
-    *mutableCookie = ConvertTo<TFragmentWriteCookiePtr>(TYsonString(req.cookie()));
+    *mutableCookie = ConvertTo<TSignedWriteFragmentCookiePtr>(TYsonString(req.signed_cookie()));
     if (req.has_config()) {
         mutableOptions->Config = ConvertTo<TTableWriterConfigPtr>(TYsonString(req.config()));
     } else {
diff --git a/yt/yt/client/api/rpc_proxy/helpers.h b/yt/yt/client/api/rpc_proxy/helpers.h
index dfe89178cb1..9e7f432e62b 100644
--- a/yt/yt/client/api/rpc_proxy/helpers.h
+++ b/yt/yt/client/api/rpc_proxy/helpers.h
@@ -296,11 +296,11 @@ void ParseRequest(
 
 void FillRequest(
     TReqFinishDistributedWriteSession* req,
-    TDistributedWriteSessionPtr session,
+    const TDistributedWriteSessionWithResults& sessionWithResults,
     const TDistributedWriteSessionFinishOptions& options);
 
 void ParseRequest(
-    TDistributedWriteSessionPtr* mutableSession,
+    TDistributedWriteSessionWithResults* mutableSessionWithResults,
     TDistributedWriteSessionFinishOptions* mutableOptions,
     const TReqFinishDistributedWriteSession& req);
 
@@ -308,12 +308,12 @@ void ParseRequest(
 
 void FillRequest(
     TReqWriteTableFragment* req,
-    const TFragmentWriteCookiePtr& cookie,
-    const TFragmentTableWriterOptions& options);
+    const TSignedWriteFragmentCookiePtr& cookie,
+    const TTableFragmentWriterOptions& options);
 
 void ParseRequest(
-    TFragmentWriteCookiePtr* mutableCookie,
-    TFragmentTableWriterOptions* mutableOptions,
+    TSignedWriteFragmentCookiePtr* mutableCookie,
+    TTableFragmentWriterOptions* mutableOptions,
     const TReqWriteTableFragment& req);
 
 } // namespace NProto
diff --git a/yt/yt/client/api/rpc_proxy/table_writer.cpp b/yt/yt/client/api/rpc_proxy/table_writer.cpp
index d957c670234..7c258019d50 100644
--- a/yt/yt/client/api/rpc_proxy/table_writer.cpp
+++ b/yt/yt/client/api/rpc_proxy/table_writer.cpp
@@ -42,4 +42,52 @@ ITableWriterPtr CreateTableWriter(
 
 ////////////////////////////////////////////////////////////////////////////////
 
+class TTableFragmentWriter
+    : public TRowBatchWriter
+    , public ITableFragmentWriter
+{
+public:
+    TTableFragmentWriter(
+        IAsyncZeroCopyOutputStreamPtr underlying,
+        TTableSchemaPtr schema,
+        TFuture<TSignedWriteFragmentResultPtr> asyncResult)
+        : TRowBatchWriter(std::move(underlying))
+        , Schema_(std::move(schema))
+        , AsyncResult_(std::move(asyncResult))
+    {
+    }
+
+    const TTableSchemaPtr& GetSchema() const override
+    {
+        return Schema_;
+    }
+
+    TFuture<void> Close() override
+    {
+        return AllSucceeded<void>({TRowBatchWriter::Close(), AsyncResult_.AsVoid()});
+    }
+
+    TSignedWriteFragmentResultPtr GetWriteFragmentResult() const override
+    {
+        YT_VERIFY(AsyncResult_.IsSet());
+        auto resultOrError = AsyncResult_.Get();
+        YT_VERIFY(resultOrError.IsOK());
+        return resultOrError.Value();
+    }
+
+private:
+    const TTableSchemaPtr Schema_;
+    const TFuture<TSignedWriteFragmentResultPtr> AsyncResult_;
+};
+
+ITableFragmentWriterPtr CreateTableFragmentWriter(
+    NConcurrency::IAsyncZeroCopyOutputStreamPtr outputStream,
+    TTableSchemaPtr tableSchema,
+    TFuture<TSignedWriteFragmentResultPtr> asyncResult)
+{
+    return New<TTableFragmentWriter>(std::move(outputStream), std::move(tableSchema), std::move(asyncResult));
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NApi::NRpcProxy
diff --git a/yt/yt/client/api/rpc_proxy/table_writer.h b/yt/yt/client/api/rpc_proxy/table_writer.h
index 21e20c4ba4b..4d8348f9899 100644
--- a/yt/yt/client/api/rpc_proxy/table_writer.h
+++ b/yt/yt/client/api/rpc_proxy/table_writer.h
@@ -12,6 +12,15 @@ ITableWriterPtr CreateTableWriter(
     NConcurrency::IAsyncZeroCopyOutputStreamPtr outputStream,
     NTableClient::TTableSchemaPtr tableSchema);
 
+// NB(arkady-e1ppa): Promise is expected to be set
+// whenever outputStream is closed.
+// TODO(arkady-e1ppa): Introduce a separate output stream
+// which has this logic built in to make this code less fragile.
+ITableFragmentWriterPtr CreateTableFragmentWriter(
+    NConcurrency::IAsyncZeroCopyOutputStreamPtr outputStream,
+    NTableClient::TTableSchemaPtr tableSchema,
+    TFuture<NTableClient::TSignedWriteFragmentResultPtr> asyncResult);
+
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NApi::NRpcProxy
diff --git a/yt/yt/client/api/rpc_proxy/transaction_impl.cpp b/yt/yt/client/api/rpc_proxy/transaction_impl.cpp
index e52fdf7acbb..89da586fdf7 100644
--- a/yt/yt/client/api/rpc_proxy/transaction_impl.cpp
+++ b/yt/yt/client/api/rpc_proxy/transaction_impl.cpp
@@ -921,7 +921,7 @@ IJournalWriterPtr TTransaction::CreateJournalWriter(
         PatchTransactionId(options));
 }
 
-TFuture<TDistributedWriteSessionPtr> TTransaction::StartDistributedWriteSession(
+TFuture<TDistributedWriteSessionWithCookies> TTransaction::StartDistributedWriteSession(
     const NYPath::TRichYPath& path,
     const TDistributedWriteSessionStartOptions& options)
 {
@@ -932,12 +932,12 @@ TFuture<TDistributedWriteSessionPtr> TTransaction::StartDistributedWriteSession(
 }
 
 TFuture<void> TTransaction::FinishDistributedWriteSession(
-    TDistributedWriteSessionPtr session,
+    const TDistributedWriteSessionWithResults& sessionWithResults,
     const TDistributedWriteSessionFinishOptions& options)
 {
     ValidateActive();
     return Client_->FinishDistributedWriteSession(
-        std::move(session),
+        sessionWithResults,
         options);
 }
 
diff --git a/yt/yt/client/api/rpc_proxy/transaction_impl.h b/yt/yt/client/api/rpc_proxy/transaction_impl.h
index 3faa58d7087..20ac689bcb2 100644
--- a/yt/yt/client/api/rpc_proxy/transaction_impl.h
+++ b/yt/yt/client/api/rpc_proxy/transaction_impl.h
@@ -235,12 +235,12 @@ public:
         const NYPath::TYPath& path,
         const NApi::TJournalWriterOptions& options) override;
 
-    TFuture<TDistributedWriteSessionPtr> StartDistributedWriteSession(
+    TFuture<TDistributedWriteSessionWithCookies> StartDistributedWriteSession(
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options = {}) override;
 
     TFuture<void> FinishDistributedWriteSession(
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options = {}) override;
 
     // Custom methods.
diff --git a/yt/yt/client/api/table_writer.h b/yt/yt/client/api/table_writer.h
index f4d42c3c78e..4ac0fa0fd79 100644
--- a/yt/yt/client/api/table_writer.h
+++ b/yt/yt/client/api/table_writer.h
@@ -18,4 +18,15 @@ DEFINE_REFCOUNTED_TYPE(ITableWriter)
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct ITableFragmentWriter
+    : public ITableWriter
+{
+    //! Returns signed write result. Only safe to use after |Close|.
+    virtual TSignedWriteFragmentResultPtr GetWriteFragmentResult() const = 0;
+};
+
+DEFINE_REFCOUNTED_TYPE(ITableFragmentWriter)
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NApi
diff --git a/yt/yt/client/driver/distributed_table_commands.cpp b/yt/yt/client/driver/distributed_table_commands.cpp
index 7bd88c5279a..bfb81c20c35 100644
--- a/yt/yt/client/driver/distributed_table_commands.cpp
+++ b/yt/yt/client/driver/distributed_table_commands.cpp
@@ -3,10 +3,13 @@
 #include "helpers.h"
 
 #include <yt/yt/client/api/distributed_table_session.h>
+#include <yt/yt/client/api/table_writer.h>
 
 #include <yt/yt/client/formats/config.h>
 #include <yt/yt/client/formats/parser.h>
 
+#include <yt/yt/client/signature/signature.h>
+
 #include <yt/yt/client/ypath/public.h>
 
 #include <yt/yt/library/formats/format.h>
@@ -37,12 +40,12 @@ void TStartDistributedWriteSessionCommand::DoExecute(ICommandContextPtr context)
 {
     auto transaction = AttachTransaction(context, /*required*/ false);
 
-    auto session = WaitFor(context->GetClient()->StartDistributedWriteSession(
+    auto sessionAndCookies = WaitFor(context->GetClient()->StartDistributedWriteSession(
         Path,
         Options));
 
-    ProduceOutput(context, [session = std::move(session)] (IYsonConsumer* consumer) {
-        Serialize(session, consumer);
+    ProduceOutput(context, [sessionAndCookies = std::move(sessionAndCookies)] (IYsonConsumer* consumer) {
+        Serialize(sessionAndCookies, consumer);
     });
 }
 
@@ -51,15 +54,20 @@ void TStartDistributedWriteSessionCommand::DoExecute(ICommandContextPtr context)
 void TFinishDistributedWriteSessionCommand::Register(TRegistrar registrar)
 {
     registrar.Parameter("session", &TThis::Session);
+    registrar.Parameter("results", &TThis::Results);
 }
 
 // -> Nothing
 void TFinishDistributedWriteSessionCommand::DoExecute(ICommandContextPtr context)
 {
-    auto session = ConvertTo<TDistributedWriteSessionPtr>(Session);
+    auto session = ConvertTo<TSignedDistributedWriteSessionPtr>(Session);
+    auto results = ConvertTo<std::vector<NTableClient::TSignedWriteFragmentResultPtr>>(Results);
 
     WaitFor(context->GetClient()->FinishDistributedWriteSession(
-        std::move(session),
+        TDistributedWriteSessionWithResults{
+            .Session = std::move(session),
+            .Results = std::move(results),
+        },
         Options))
             .ThrowOnError();
 
@@ -70,33 +78,43 @@ void TFinishDistributedWriteSessionCommand::DoExecute(ICommandContextPtr context
 
 void TWriteTableFragmentCommand::Execute(ICommandContextPtr context)
 {
-    TTypedCommand<NApi::TFragmentTableWriterOptions>::Execute(std::move(context));
+    TTypedCommand<NApi::TTableFragmentWriterOptions>::Execute(std::move(context));
 }
 
-void TWriteTableFragmentCommand::Register(TRegistrar /*registrar*/)
-{ }
+void TWriteTableFragmentCommand::Register(TRegistrar registrar)
+{
+    registrar.Parameter("cookie", &TThis::Cookie);
+}
 
-TFuture<NApi::ITableWriterPtr> TWriteTableFragmentCommand::CreateTableWriter(
-    const ICommandContextPtr& context) const
+NApi::ITableWriterPtr TWriteTableFragmentCommand::CreateTableWriter(
+    const ICommandContextPtr& context)
 {
-    PutMethodInfoInTraceContext("participant_write_table");
+    PutMethodInfoInTraceContext("write_table_fragment");
 
-    return context
+    auto tableWriter = WaitFor(context
         ->GetClient()
-        ->CreateFragmentTableWriter(
-            StaticPointerCast<TFragmentWriteCookie>(ResultingCookie),
-            TTypedCommand<TFragmentTableWriterOptions>::Options);
+        ->CreateTableFragmentWriter(
+            ConvertTo<TSignedWriteFragmentCookiePtr>(Cookie),
+            TTypedCommand<TTableFragmentWriterOptions>::Options))
+                .ValueOrThrow();
+
+    TableWriter = tableWriter;
+    return tableWriter;
 }
 
 // -> Cookie
 void TWriteTableFragmentCommand::DoExecute(ICommandContextPtr context)
 {
-    auto cookie = ConvertTo<TFragmentWriteCookiePtr>(Cookie);
-    ResultingCookie = StaticPointerCast<TRefCounted>(std::move(cookie));
+    auto cookie = ConvertTo<TSignedWriteFragmentCookiePtr>(Cookie);
 
     DoExecuteImpl(context);
-    ProduceOutput(context, [cookie = std::move(ResultingCookie)] (IYsonConsumer* consumer) {
-        Serialize(StaticPointerCast<TFragmentWriteCookie>(cookie), consumer);
+
+    // Sadly, we are plagued by virtual bases :/.
+    auto writer = DynamicPointerCast<NApi::ITableFragmentWriter>(TableWriter);
+    ProduceOutput(context, [result = writer->GetWriteFragmentResult()] (IYsonConsumer* consumer) {
+        Serialize(
+            *result.Underlying(),
+            consumer);
     });
 }
 
diff --git a/yt/yt/client/driver/distributed_table_commands.h b/yt/yt/client/driver/distributed_table_commands.h
index fa3d6bb4e53..57fcfd7c6df 100644
--- a/yt/yt/client/driver/distributed_table_commands.h
+++ b/yt/yt/client/driver/distributed_table_commands.h
@@ -38,6 +38,7 @@ public:
 
 private:
     NYTree::INodePtr Session;
+    std::vector<NYTree::INodePtr> Results;
 
     void DoExecute(ICommandContextPtr context) override;
 };
@@ -46,7 +47,7 @@ private:
 
 // -> Cookie
 class TWriteTableFragmentCommand
-    : public TTypedCommand<NApi::TFragmentTableWriterOptions>
+    : public TTypedCommand<NApi::TTableFragmentWriterOptions>
     , private TWriteTableCommand
 {
 public:
@@ -62,10 +63,10 @@ private:
     using TBase = TWriteTableCommand;
 
     NYTree::INodePtr Cookie;
-    TRefCountedPtr ResultingCookie;
+    TRefCountedPtr TableWriter;
 
-    TFuture<NApi::ITableWriterPtr> CreateTableWriter(
-        const ICommandContextPtr& context) const override;
+    NApi::ITableWriterPtr CreateTableWriter(
+        const ICommandContextPtr& context) override;
 
     void DoExecute(ICommandContextPtr context) override;
 };
diff --git a/yt/yt/client/driver/driver.cpp b/yt/yt/client/driver/driver.cpp
index 32568fdc3cb..9be3e37f3ec 100644
--- a/yt/yt/client/driver/driver.cpp
+++ b/yt/yt/client/driver/driver.cpp
@@ -397,6 +397,8 @@ public:
             REGISTER_ALL(TRevokeLeaseCommand,              "revoke_lease",                    Null,       Structured, true,  false);
             REGISTER_ALL(TReferenceLeaseCommand,           "reference_lease",                 Null,       Structured, true,  false);
             REGISTER_ALL(TUnreferenceLeaseCommand,         "unreference_lease",               Null,       Structured, true,  false);
+
+            // TODO(arkady-e1ppa): flags past command name might be complete rubbish -- think them through later.
             REGISTER_ALL(TStartDistributedWriteSessionCommand,    "start_distributed_write_session",      Null,    Structured, true, false);
             REGISTER_ALL(TFinishDistributedWriteSessionCommand,   "finish_distributed_write_session",     Null,    Null,       true, false);
             REGISTER_ALL(TWriteTableFragmentCommand,               "distributed_write_table_partition",    Tabular, Structured, true, true );
diff --git a/yt/yt/client/driver/table_commands.cpp b/yt/yt/client/driver/table_commands.cpp
index 3ae3e78a234..e358d2f8ee6 100644
--- a/yt/yt/client/driver/table_commands.cpp
+++ b/yt/yt/client/driver/table_commands.cpp
@@ -267,14 +267,15 @@ void TWriteTableCommand::Register(TRegistrar registrar)
         .Default(1_MB);
 }
 
-TFuture<ITableWriterPtr> TWriteTableCommand::CreateTableWriter(
-    const ICommandContextPtr& context) const
+NApi::ITableWriterPtr TWriteTableCommand::CreateTableWriter(
+    const ICommandContextPtr& context)
 {
     PutMethodInfoInTraceContext("write_table");
 
-    return context->GetClient()->CreateTableWriter(
+    return WaitFor(context->GetClient()->CreateTableWriter(
         Path,
-        Options);
+        Options))
+            .ValueOrThrow();
 }
 
 void TWriteTableCommand::DoExecuteImpl(const ICommandContextPtr& context)
@@ -289,8 +290,7 @@ void TWriteTableCommand::DoExecuteImpl(const ICommandContextPtr& context)
     Options.PingAncestors = true;
     Options.Config = config;
 
-    auto apiWriter = WaitFor(CreateTableWriter(context))
-        .ValueOrThrow();
+    auto apiWriter = CreateTableWriter(context);
 
     auto schemalessWriter = CreateSchemalessFromApiWriterAdapter(std::move(apiWriter));
 
diff --git a/yt/yt/client/driver/table_commands.h b/yt/yt/client/driver/table_commands.h
index d511e766221..97f52e44fd3 100644
--- a/yt/yt/client/driver/table_commands.h
+++ b/yt/yt/client/driver/table_commands.h
@@ -83,8 +83,8 @@ public:
     static void Register(TRegistrar registrar);
 
 protected:
-    virtual TFuture<NApi::ITableWriterPtr> CreateTableWriter(
-        const ICommandContextPtr& context) const;
+    virtual NApi::ITableWriterPtr CreateTableWriter(
+        const ICommandContextPtr& context);
 
     void DoExecuteImpl(const ICommandContextPtr& context);
 
diff --git a/yt/yt/client/federated/client.cpp b/yt/yt/client/federated/client.cpp
index 170dedf9105..86555b39d62 100644
--- a/yt/yt/client/federated/client.cpp
+++ b/yt/yt/client/federated/client.cpp
@@ -236,8 +236,8 @@ public:
     UNIMPLEMENTED_METHOD(IFileWriterPtr, CreateFileWriter, (const NYPath::TRichYPath&, const TFileWriterOptions&));
     UNIMPLEMENTED_METHOD(IJournalReaderPtr, CreateJournalReader, (const NYPath::TYPath&, const TJournalReaderOptions&));
     UNIMPLEMENTED_METHOD(IJournalWriterPtr, CreateJournalWriter, (const NYPath::TYPath&, const TJournalWriterOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (TDistributedWriteSessionPtr, const TDistributedWriteSessionFinishOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (const TDistributedWriteSessionWithResults&, const TDistributedWriteSessionFinishOptions&));
 
 private:
     const TClientPtr Client_;
@@ -484,9 +484,9 @@ public:
     UNIMPLEMENTED_METHOD(TFuture<void>, PausePipeline, (const NYPath::TYPath&, const TPausePipelineOptions&));
     UNIMPLEMENTED_METHOD(TFuture<TPipelineState>, GetPipelineState, (const NYPath::TYPath&, const TGetPipelineStateOptions&));
     UNIMPLEMENTED_METHOD(TFuture<TGetFlowViewResult>, GetFlowView, (const NYPath::TYPath&, const NYPath::TYPath&, const TGetFlowViewOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (TDistributedWriteSessionPtr, const TDistributedWriteSessionFinishOptions&));
-    UNIMPLEMENTED_METHOD(TFuture<ITableWriterPtr>, CreateFragmentTableWriter, (const TFragmentWriteCookiePtr&, const TFragmentTableWriterOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (const TDistributedWriteSessionWithResults&, const TDistributedWriteSessionFinishOptions&));
+    UNIMPLEMENTED_METHOD(TFuture<ITableFragmentWriterPtr>, CreateTableFragmentWriter, (const TSignedWriteFragmentCookiePtr&, const TTableFragmentWriterOptions&));
     UNIMPLEMENTED_METHOD(TFuture<TShuffleHandlePtr>, StartShuffle, (const std::string& , int, NObjectClient::TTransactionId, const TStartShuffleOptions&));
     UNIMPLEMENTED_METHOD(TFuture<IRowBatchReaderPtr>, CreateShuffleReader, (const TShuffleHandlePtr&, int, const NTableClient::TTableReaderConfigPtr&));
     UNIMPLEMENTED_METHOD(TFuture<IRowBatchWriterPtr>, CreateShuffleWriter, (const TShuffleHandlePtr&, const std::string&, const NTableClient::TTableWriterConfigPtr&));
diff --git a/yt/yt/client/hedging/hedging.cpp b/yt/yt/client/hedging/hedging.cpp
index ac96c64e391..e5bdd74b228 100644
--- a/yt/yt/client/hedging/hedging.cpp
+++ b/yt/yt/client/hedging/hedging.cpp
@@ -109,9 +109,9 @@ public:
     UNSUPPORTED_METHOD(IFileWriterPtr, CreateFileWriter, (const TRichYPath&, const TFileWriterOptions&));
     UNSUPPORTED_METHOD(IJournalReaderPtr, CreateJournalReader, (const TYPath&, const TJournalReaderOptions&));
     UNSUPPORTED_METHOD(IJournalWriterPtr, CreateJournalWriter, (const TYPath&, const TJournalWriterOptions&));
-    UNSUPPORTED_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
-    UNSUPPORTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (TDistributedWriteSessionPtr, const TDistributedWriteSessionFinishOptions&));
-    UNSUPPORTED_METHOD(TFuture<ITableWriterPtr>, CreateFragmentTableWriter, (const TFragmentWriteCookiePtr&, const TFragmentTableWriterOptions&));
+    UNSUPPORTED_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (const NYPath::TRichYPath&, const TDistributedWriteSessionStartOptions&));
+    UNSUPPORTED_METHOD(TFuture<void>, FinishDistributedWriteSession, (const TDistributedWriteSessionWithResults&, const TDistributedWriteSessionFinishOptions&));
+    UNSUPPORTED_METHOD(TFuture<ITableFragmentWriterPtr>, CreateTableFragmentWriter, (const TSignedWriteFragmentCookiePtr&, const TTableFragmentWriterOptions&));
 
     // IClient methods.
     // Unsupported methods.
diff --git a/yt/yt/client/signature/signature.h b/yt/yt/client/signature/signature.h
index b163ecc0e93..6cb739bd3c0 100644
--- a/yt/yt/client/signature/signature.h
+++ b/yt/yt/client/signature/signature.h
@@ -26,7 +26,9 @@ public:
     [[nodiscard]] const NYson::TYsonString& Payload() const;
 
 private:
-    NYson::TYsonString Header_;
+    // TODO(arkady-e1ppa): Whenever trivial generator/validators are added
+    // remove initialization.
+    NYson::TYsonString Header_ = NYson::TYsonString(TStringBuf(""));
     NYson::TYsonString Payload_;
     std::vector<std::byte> Signature_;
 
diff --git a/yt/yt/client/table_client/adapters.cpp b/yt/yt/client/table_client/adapters.cpp
index 648a208ea43..f17c1dcdac9 100644
--- a/yt/yt/client/table_client/adapters.cpp
+++ b/yt/yt/client/table_client/adapters.cpp
@@ -25,47 +25,69 @@ static constexpr auto& Logger = TableClientLogger;
 
 ////////////////////////////////////////////////////////////////////////////////
 
+template <std::derived_from<ITableWriter> TInterface, std::derived_from<IUnversionedWriter> TUnderlyingInterface>
 class TApiFromSchemalessWriterAdapter
-    : public ITableWriter
+    : public TInterface
 {
 public:
-    explicit TApiFromSchemalessWriterAdapter(IUnversionedWriterPtr underlyingWriter)
+    explicit TApiFromSchemalessWriterAdapter(TIntrusivePtr<TUnderlyingInterface> underlyingWriter)
         : UnderlyingWriter_(std::move(underlyingWriter))
     { }
 
-    bool Write(TRange<TUnversionedRow> rows) override
+    bool Write(TRange<TUnversionedRow> rows) /*override*/
     {
         return UnderlyingWriter_->Write(rows);
     }
 
-    TFuture<void> GetReadyEvent() override
+    TFuture<void> GetReadyEvent() /*override*/
     {
         return UnderlyingWriter_->GetReadyEvent();
     }
 
-    TFuture<void> Close() override
+    TFuture<void> Close() /*override*/
     {
         return UnderlyingWriter_->Close();
     }
 
-    const TNameTablePtr& GetNameTable() const override
+    const TNameTablePtr& GetNameTable() const /*override*/
     {
         return UnderlyingWriter_->GetNameTable();
     }
 
-    const TTableSchemaPtr& GetSchema() const override
+    const TTableSchemaPtr& GetSchema() const /*override*/
     {
         return UnderlyingWriter_->GetSchema();
     }
 
-private:
-    const IUnversionedWriterPtr UnderlyingWriter_;
+protected:
+    const TIntrusivePtr<TUnderlyingInterface> UnderlyingWriter_;
 };
 
 ITableWriterPtr CreateApiFromSchemalessWriterAdapter(
     IUnversionedWriterPtr underlyingWriter)
 {
-    return New<TApiFromSchemalessWriterAdapter>(std::move(underlyingWriter));
+    return New<TApiFromSchemalessWriterAdapter<ITableWriter, IUnversionedWriter>>(std::move(underlyingWriter));
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TApiFromSchemalessTableFragmentWriterAdapter
+    : public TApiFromSchemalessWriterAdapter<ITableFragmentWriter, IUnversionedTableFragmentWriter>
+{
+public:
+    using TBase = TApiFromSchemalessWriterAdapter<ITableFragmentWriter, IUnversionedTableFragmentWriter>;
+    using TBase::TBase;
+
+    TSignedWriteFragmentResultPtr GetWriteFragmentResult() const /*override*/
+    {
+        return TBase::UnderlyingWriter_->GetWriteFragmentResult();
+    }
+};
+
+ITableFragmentWriterPtr CreateApiFromSchemalessWriterAdapter(
+    IUnversionedTableFragmentWriterPtr underlyingWriter)
+{
+    return New<TApiFromSchemalessTableFragmentWriterAdapter>(std::move(underlyingWriter));
 }
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/client/table_client/adapters.h b/yt/yt/client/table_client/adapters.h
index f90cbe101d0..f4b058bf23c 100644
--- a/yt/yt/client/table_client/adapters.h
+++ b/yt/yt/client/table_client/adapters.h
@@ -21,6 +21,9 @@ IUnversionedWriterPtr CreateSchemalessFromApiWriterAdapter(
 NApi::ITableWriterPtr CreateApiFromSchemalessWriterAdapter(
     IUnversionedWriterPtr underlyingWriter);
 
+NApi::ITableFragmentWriterPtr CreateApiFromSchemalessWriterAdapter(
+    IUnversionedTableFragmentWriterPtr underlyingWriter);
+
 ////////////////////////////////////////////////////////////////////////////////
 
 struct TPipeReaderToWriterOptions
diff --git a/yt/yt/client/table_client/public.h b/yt/yt/client/table_client/public.h
index b577fb929db..c397838e60f 100644
--- a/yt/yt/client/table_client/public.h
+++ b/yt/yt/client/table_client/public.h
@@ -4,6 +4,8 @@
 
 #include <yt/yt/client/cypress_client/public.h>
 
+#include <yt/yt/client/signature/public.h>
+
 #include <yt/yt/client/tablet_client/public.h>
 
 #include <yt/yt/client/transaction_client/public.h>
@@ -343,6 +345,7 @@ DECLARE_REFCOUNTED_CLASS(TRowBuffer)
 DECLARE_REFCOUNTED_STRUCT(ISchemalessUnversionedReader)
 DECLARE_REFCOUNTED_STRUCT(ISchemafulUnversionedReader)
 DECLARE_REFCOUNTED_STRUCT(IUnversionedWriter)
+DECLARE_REFCOUNTED_STRUCT(IUnversionedTableFragmentWriter)
 DECLARE_REFCOUNTED_STRUCT(IUnversionedRowsetWriter)
 
 using TSchemalessWriterFactory = std::function<IUnversionedRowsetWriterPtr(
@@ -454,4 +457,10 @@ struct TVersionedWriteOptions;
 
 ////////////////////////////////////////////////////////////////////////////////
 
+YT_DEFINE_STRONG_TYPEDEF(TSignedDistributedWriteSessionPtr, NSignature::TSignaturePtr);
+YT_DEFINE_STRONG_TYPEDEF(TSignedWriteFragmentCookiePtr, NSignature::TSignaturePtr);
+YT_DEFINE_STRONG_TYPEDEF(TSignedWriteFragmentResultPtr, NSignature::TSignaturePtr);
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NTableClient
diff --git a/yt/yt/client/table_client/unversioned_writer.h b/yt/yt/client/table_client/unversioned_writer.h
index b8ba78d6214..22814d3b0b1 100644
--- a/yt/yt/client/table_client/unversioned_writer.h
+++ b/yt/yt/client/table_client/unversioned_writer.h
@@ -62,4 +62,15 @@ DEFINE_REFCOUNTED_TYPE(IUnversionedWriter)
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct IUnversionedTableFragmentWriter
+    : public IUnversionedWriter
+{
+    //! Returns signed write result. Only safe to use after |Stop|.
+    virtual TSignedWriteFragmentResultPtr GetWriteFragmentResult() const = 0;
+};
+
+DEFINE_REFCOUNTED_TYPE(IUnversionedTableFragmentWriter)
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NTableClient
diff --git a/yt/yt/client/unittests/mock/client.h b/yt/yt/client/unittests/mock/client.h
index 31c76a9071a..e29d989ceb5 100644
--- a/yt/yt/client/unittests/mock/client.h
+++ b/yt/yt/client/unittests/mock/client.h
@@ -831,19 +831,19 @@ public:
         const TGetFlowViewOptions& options),
         (override));
 
-    MOCK_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (
+    MOCK_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options),
         (override));
 
     MOCK_METHOD(TFuture<void>, FinishDistributedWriteSession, (
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options),
         (override));
 
-    MOCK_METHOD(TFuture<ITableWriterPtr>, CreateFragmentTableWriter, (
-        const TFragmentWriteCookiePtr& cookie,
-        const TFragmentTableWriterOptions& options),
+    MOCK_METHOD(TFuture<ITableFragmentWriterPtr>, CreateTableFragmentWriter, (
+        const TSignedWriteFragmentCookiePtr& cookie,
+        const TTableFragmentWriterOptions& options),
         (override));
 
     MOCK_METHOD(TFuture<TShuffleHandlePtr>, StartShuffle, (
diff --git a/yt/yt/client/unittests/mock/transaction.h b/yt/yt/client/unittests/mock/transaction.h
index 943e4fa2015..713bf0b4430 100644
--- a/yt/yt/client/unittests/mock/transaction.h
+++ b/yt/yt/client/unittests/mock/transaction.h
@@ -151,13 +151,13 @@ public:
         const NYPath::TYPath& path,
         const TJournalWriterOptions& options), (override));
 
-    MOCK_METHOD(TFuture<TDistributedWriteSessionPtr>, StartDistributedWriteSession, (
+    MOCK_METHOD(TFuture<TDistributedWriteSessionWithCookies>, StartDistributedWriteSession, (
         const NYPath::TRichYPath& path,
         const TDistributedWriteSessionStartOptions& options),
         (override));
 
     MOCK_METHOD(TFuture<void>, FinishDistributedWriteSession, (
-        TDistributedWriteSessionPtr session,
+        const TDistributedWriteSessionWithResults& sessionWithResults,
         const TDistributedWriteSessionFinishOptions& options),
         (override));
 
diff --git a/yt/yt/core/concurrency/throughput_throttler.cpp b/yt/yt/core/concurrency/throughput_throttler.cpp
index c5a4e34ce67..51518b1cd48 100644
--- a/yt/yt/core/concurrency/throughput_throttler.cpp
+++ b/yt/yt/core/concurrency/throughput_throttler.cpp
@@ -448,7 +448,8 @@ private:
         YT_VERIFY(limit >= 0);
 
         // Reconfigure clears the update cookie, so infinity is fine.
-        auto delay = limit ? TDuration::MilliSeconds(Max<i64>(0, -Available_ * 1000 / limit)) : TDuration::Max();
+        // NB: Cap by 1 ms from below to somewhat limit the wakeup rate.
+        auto delay = limit ? TDuration::MilliSeconds(Max<i64>(1, -Available_ * 1000 / limit)) : TDuration::Max();
 
         UpdateCookie_ = TDelayedExecutor::Submit(
             BIND_NO_PROPAGATE(&TReconfigurableThroughputThrottler::Update, MakeWeak(this)),
diff --git a/yt/yt/core/rpc/stream.h b/yt/yt/core/rpc/stream.h
index f2220e74253..c7d765c88e1 100644
--- a/yt/yt/core/rpc/stream.h
+++ b/yt/yt/core/rpc/stream.h
@@ -266,6 +266,9 @@ TFuture<NConcurrency::IAsyncZeroCopyOutputStreamPtr> CreateRpcClientOutputStream
     TCallback<void(TSharedRef)> metaHandler);
 
 //! This variant additionally allows non-trivial response of streaming request to be handled.
+//! TODO(arkady-e1ppa): Introduce IAsyncZeroCopyOutputStream<TRet> which |Close| returns
+//! TFuture<TRet> instead of TFuture<void> as a way to transfer data via rsp
+//! use it here.
 template <class TRequestMessage, class TResponse>
 TFuture<NConcurrency::IAsyncZeroCopyOutputStreamPtr> CreateRpcClientOutputStream(
     TIntrusivePtr<TTypedClientRequest<TRequestMessage, TResponse>> request,
diff --git a/yt/yt/core/ytree/yson_struct-inl.h b/yt/yt/core/ytree/yson_struct-inl.h
index aad0bd1fd0b..6f56b6bd48c 100644
--- a/yt/yt/core/ytree/yson_struct-inl.h
+++ b/yt/yt/core/ytree/yson_struct-inl.h
@@ -331,13 +331,14 @@ std::vector<TIntrusivePtr<T>> CloneYsonStructs(const std::vector<TIntrusivePtr<T
     return clonedObjs;
 }
 
-template <class T>
-THashMap<TString, TIntrusivePtr<T>> CloneYsonStructs(const THashMap<TString, TIntrusivePtr<T>>& objs)
+template <class TKey, class TValue>
+THashMap<TKey, TIntrusivePtr<TValue>> CloneYsonStructs(const THashMap<TKey, TIntrusivePtr<TValue>>& objs)
 {
-    THashMap<TString, TIntrusivePtr<T>> clonedObjs;
+    THashMap<TKey, TIntrusivePtr<TValue>> clonedObjs;
     clonedObjs.reserve(objs.size());
     for (const auto& [key, obj] : objs) {
-        clonedObjs.emplace(key, CloneYsonStruct(obj));
+        // TODO(babenko): switch to std::string
+        clonedObjs.emplace(TString(key), CloneYsonStruct(obj));
     }
     return clonedObjs;
 }
diff --git a/yt/yt/core/ytree/yson_struct.h b/yt/yt/core/ytree/yson_struct.h
index 202dbdbe033..bc42b7073ec 100644
--- a/yt/yt/core/ytree/yson_struct.h
+++ b/yt/yt/core/ytree/yson_struct.h
@@ -383,8 +383,8 @@ template <class T>
 TIntrusivePtr<T> CloneYsonStruct(const TIntrusivePtr<T>& obj, bool postprocess = true, bool setDefaults = true);
 template <class T>
 std::vector<TIntrusivePtr<T>> CloneYsonStructs(const std::vector<TIntrusivePtr<T>>& objs);
-template <class T>
-THashMap<TString, TIntrusivePtr<T>> CloneYsonStructs(const THashMap<TString, TIntrusivePtr<T>>& objs);
+template <class TKey, class TValue>
+THashMap<TKey, TIntrusivePtr<TValue>> CloneYsonStructs(const THashMap<TKey, TIntrusivePtr<TValue>>& objs);
 
 void Serialize(const TYsonStructBase& value, NYson::IYsonConsumer* consumer);
 void Deserialize(TYsonStructBase& value, INodePtr node);
diff --git a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
index 723a63b639f..457a28b0f58 100644
--- a/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
+++ b/yt/yt_proto/yt/client/api/rpc_proxy/proto/api_service.proto
@@ -3426,20 +3426,23 @@ message TRspGetQueryTrackerInfo
 message TReqStartDistributedWriteSession
 {
     required string path = 1;
+    required int32 cookie_count = 2;
 
     optional TTransactionalOptions transactional_options = 100;
 }
 
 message TRspStartDistributedWriteSession
 {
-    required bytes session = 1; // YSON-serialized TDistributedWriteSession
+    required bytes signed_session = 1; // YSON-serialized TSignedDistributedWriteSessionPtr
+    repeated bytes signed_cookies = 2; // vector of YSON-serialized TSignedDistributedWriteCookiePtr
 }
 
 message TReqFinishDistributedWriteSession
 {
-    required bytes session = 1; // YSON-serialized TDistributedWriteSession
+    required bytes signed_session = 1; // YSON-serialized TSignedDistributedWriteSessionPtr
+    repeated bytes signed_write_results = 2; // vector of YSON-serialized TSignedWriteFragmentResultPtr
 
-    required int32 max_children_per_attach_request = 2;
+    required int32 max_children_per_attach_request = 3;
 }
 
 message TRspFinishDistributedWriteSession
@@ -3452,12 +3455,12 @@ message TReqWriteTableFragment
 
     optional bytes format = 2; // YSON-serialized TFormat
 
-    required bytes cookie = 3; // YSON-serialized TFragmentWriteCookie
+    required bytes signed_cookie = 3; // YSON-serialized TSignedWriteFragmentCookiePtr
 }
 
 message TRspWriteTableFragment
 {
-    required bytes cookie = 1; // YSON-serialized TFragmentWriteCookie
+    required bytes signed_write_result = 1; // YSON-serialized TSignedWriteFragmentResultPtr
 }
 
 ///////////////////////////////////////////////////////////////////////////////