diff --git a/docs/guides/profiler-support.md b/docs/guides/profiler-support.md
index 6246bd5..f2fa4b3 100644
--- a/docs/guides/profiler-support.md
+++ b/docs/guides/profiler-support.md
@@ -58,6 +58,47 @@ bk_profiler <tool> [options] -- <command ...>
 `bk_resolve_profiler_tool` / `bk_resolve_profiler_level` を使う。既存運用や site-local
 調整のために app 固有変数が必要な場合は、共通変数の上書きとして helper に渡す。
 
+## Workflow Stage Timing
+
+`bk_profiler` は各 `fapp` / `ncu` acquisition と report export の時間を自動記録する。
+アプリの `profile.sh` に開始・終了時計、JSON生成、artifact登録を実装する必要はない。
+SBDとGENESISも同じ共通経路を使い、MPI rank選択やcontainer起動条件はアプリ側で組み立てる。
+
+`run.sh` では、実験条件と出力先を決めた時点で共通実行contextを設定する。
+contextはsubshellやpipelineへ引き継がれ、作業ディレクトリを変えても出力先は変わらない。
+
+```bash
+bk_run_context --results-dir "${PWD}/results" --exp "$experiment"
+bk_run --log benchmark.log -- mpirun -np "$ranks" ./application
+bk_profiler ncu --archive "${BK_RUN_RESULTS_DIR}/padata.tgz" -- ./application
+```
+
+独自のMPI/container起動が必要な場合は、組み立てた完全なコマンドを
+`bk_profile_execute --tool <tool> --phase <collect|export|plan> -- <command ...>`
+へ渡す。`--profile <name>` で対象profileを識別でき、`--log <file>` でコマンドの
+stdout/stderrだけをファイルへ送れる。NSYSのcollect/exportにもこの入口を使う。
+NCU plan生成は `bk_generate_ncu_plan` へ従来のgenerator引数を渡す。
+いずれも時計操作や記録ファイルの管理は共通層が担当する。
+
+共通層は実験条件ごとの `results/workflow_timing_*.json` にstage、tool、profile、
+UTC開始・終了時刻、monotonic時計による経過秒、終了コードを保存する。
+更新はlockとatomic replacementを使い、並列コマンドの追記を保持する。
+`scripts/result.sh` が現在の実行sessionの記録を `timing_observations` へ自動追加し、
+既存のapp独自timing observationも保持する。Resultの `Exp` によりscopeを絞り、
+既存のMeasurement Artifacts送信・console表示経路を使う。公開面には追加しない。
+context未設定の `bk_profiler` はarchiveの出力ディレクトリを使い、job単位のscopeとなる。
+一つのresultsディレクトリは一つの実行sessionで使用する。
+
+記録はコマンドのwall timeであり、MPI起動やprofilerのreport finalizationを含む。
+queue待ちやarchive圧縮は含まず、アプリのFOMを置き換えない。profile用の再実行時間を
+そのままアプリのslowdown比とは解釈せず、同条件の通常実行と比較する。
+計測用helper自身の起動・記録には小さな追加コストがある。
+
+失敗・timeoutはコマンドの終了コードを保持する。finishを書けずに中断したstageは
+`running` と開始時刻だけが残るため、未完了として扱い、成功やゼロ秒とみなさない。
+任意の記録処理が失敗しても実行コマンドの終了コードを変えない。schedulerによる強制終了後の
+転送可否はjobのartifact upload方針に依存する。
+
 ## 3. 共通語彙としての level
 
 `single/simple/standard/detailed` は Benchkit の共通語彙として扱う。
diff --git a/programs/genesis/profile.sh b/programs/genesis/profile.sh
index 8ab29ab..3f24ccf 100644
--- a/programs/genesis/profile.sh
+++ b/programs/genesis/profile.sh
@@ -428,7 +428,8 @@ genesis_run_container_ncu_acquisition_profile() {
     echo "bk_run_ncu_acquisition_profile: profile='${profile_name}' kernel='${kernel_regex}' skip=${launch_skip} count=${launch_count}" >&2
     echo "bk_profiler[ncu]: starting ${rep_name} level=${profiler_level} inside container rank 0" >&2
     set +e
-    "${profile_cmd[@]}" </dev/null 2>&1 | tee "$profile_log"
+    bk_profile_execute --tool ncu --phase collect --profile "$profile_slug" -- \
+        "${profile_cmd[@]}" </dev/null 2>&1 | tee "$profile_log"
     profiler_status=${PIPESTATUS[0]}
     set -e
 
@@ -446,12 +447,14 @@ genesis_run_container_ncu_acquisition_profile() {
             --csv \
             --print-units base \
             --print-fp || return 1
-        "${import_cmd[@]}" > "${rep_dir}/profile_raw.csv" 2> "${rep_dir}/profile_raw.csv.log" || true
+        bk_profile_execute --tool ncu --phase export --profile "${profile_slug}/raw" -- \
+            "${import_cmd[@]}" > "${rep_dir}/profile_raw.csv" 2> "${rep_dir}/profile_raw.csv.log" || true
 
         import_cmd=()
         genesis_build_container_once_command app_cmd import_cmd \
             ncu --import "$report_file" --page details || return 1
-        "${import_cmd[@]}" > "$stage_dir/reports/ncu_import_${rep_name}.txt" 2>&1 || true
+        bk_profile_execute --tool ncu --phase export --profile "${profile_slug}/details" -- \
+            "${import_cmd[@]}" > "$stage_dir/reports/ncu_import_${rep_name}.txt" 2>&1 || true
     fi
 
     cp -R "$rep_dir" "$stage_dir/raw/${rep_name}"
@@ -610,9 +613,9 @@ genesis_generate_ncu_plan() {
                 "${discovery_cmd[@]:$GENESIS_APPTAINER_PAYLOAD_INDEX}"
             )
             genesis_build_container_rank0_profile_command nsys_payload discovery_cmd nsys_profile_cmd
-            "${nsys_profile_cmd[@]}" 2>&1 | tee "$nsys_log" >&2
+            bk_profile_execute --tool nsys --phase collect -- "${nsys_profile_cmd[@]}" 2>&1 | tee "$nsys_log" >&2
         else
-            nsys profile \
+            bk_profile_execute --tool nsys --phase collect -- nsys profile \
                 --force-overwrite=true \
                 --trace=cuda \
                 --sample=none \
@@ -633,10 +636,11 @@ genesis_generate_ncu_plan() {
         if genesis_find_apptainer_payload_index discovery_cmd; then
             genesis_build_container_once_command discovery_cmd nsys_stats_cmd \
                 nsys stats --force-export=true --report cuda_gpu_kern_sum --format csv --output "$nsys_csv" "$nsys_report" || return 1
-            "${nsys_stats_cmd[@]}" >/dev/null
+            bk_profile_execute --tool nsys --phase export -- "${nsys_stats_cmd[@]}" >/dev/null
             nsys_stats_status=$?
         else
-            nsys stats --force-export=true --report cuda_gpu_kern_sum --format csv --output "$nsys_csv" "$nsys_report" >/dev/null
+            bk_profile_execute --tool nsys --phase export -- \
+                nsys stats --force-export=true --report cuda_gpu_kern_sum --format csv --output "$nsys_csv" "$nsys_report" >/dev/null
             nsys_stats_status=$?
         fi
         if [ "$nsys_stats_status" -ne 0 ]; then
@@ -670,7 +674,7 @@ genesis_generate_ncu_plan() {
         esac
     fi
 
-    "$python_bin" "${SCRIPT_DIR}/scripts/profiling/generate_ncu_plan.py" \
+    bk_generate_ncu_plan \
         --nsys-csv "$discovery_csv" \
         --out-discovery "$discovery_json" \
         --out-plan "$plan_json" \
diff --git a/programs/genesis/run.sh b/programs/genesis/run.sh
index b30c012..e680744 100644
--- a/programs/genesis/run.sh
+++ b/programs/genesis/run.sh
@@ -28,6 +28,7 @@ input=${header}.inp
 resultsdir=${SCRIPT_DIR}/results
 artifactsdir=${SCRIPT_DIR}/artifacts
 mkdir -p ${resultsdir}
+bk_run_context --results-dir "${resultsdir}" --exp "$exp"
 export BK_INPUT_INFO_FILE="${resultsdir}/input_info.json"
 export BK_INPUT_INFO_ITEMS_FILE="${resultsdir}/.input_info_items.jsonl"
 bk_reset_input_info
@@ -162,7 +163,7 @@ run_genesis_nvidia_gpu() {
     genesis_configure_ncu_profile "$system_name" "$profiler_tool_var" "$profiler_level_var" "$module_var" || return 1
 
     echo "Running ${system_name} as NVIDIA GPU benchmark run without profiler"
-    "${nvidia_mpi_cmd[@]}" ./${binary} ${input}.sub 2>&1 | tee ${output}
+    bk_run -- "${nvidia_mpi_cmd[@]}" ./${binary} ${input}.sub 2>&1 | tee ${output}
     genesis_run_configured_ncu_profiles "$system_name" "${nvidia_mpi_cmd[@]}" ./${binary} ${input}.sub || return 1
 }
 
@@ -203,7 +204,9 @@ case "$system" in
     export PARALLEL=${nthreads}
     export OMP_NUM_THREADS=${nthreads}
 	echo "${mpi_cmd} ./${binary} ${input}.sub"
-	${mpi_cmd} ./${binary} ${input}.sub
+	# Intentional word splitting: mpi_cmd contains the launcher and its arguments.
+	# shellcheck disable=SC2086
+	bk_run -- ${mpi_cmd} ./${binary} ${input}.sub
 	[[ -f ./stdout.1.0 ]] && cp ./stdout.1.0 ${output}
 	[[ -f ./stderr.1.0 ]] && cp ./stderr.1.0 ${stderr}
     ;;
diff --git a/programs/qws/run.sh b/programs/qws/run.sh
index 10b9d49..f61dea9 100644
--- a/programs/qws/run.sh
+++ b/programs/qws/run.sh
@@ -109,6 +109,7 @@ case "$system" in
                 mpiexec -n 2 ./main "${qws_case1_args[@]}" > CASE1
                 print_results output.${PJM_JOBID}/0/2/stdout.2.0 CASE1 2 >> ../results/result
                 if bk_profiler_enabled "$qws_profiler_tool"; then
+                    bk_run_context --results-dir ../results --exp CASE0
                     bk_profiler "$qws_profiler_tool" --level "$qws_profiler_level" --archive ../results/padata0.tgz --raw-dir pa -- mpiexec -n 1 ./main "${qws_case0_args[@]}" > CASE0.profile
                 # else
                 #     emit_qws_dummy_padata ../results/padata0.tgz
@@ -118,6 +119,7 @@ case "$system" in
                 mpiexec -n 8 ./main "${qws_case7_args[@]}" > CASE7
                 print_results output.${PJM_JOBID}/0/1/stdout.1.0 CASE7 4 >> ../results/result
                 if bk_profiler_enabled "$qws_profiler_tool"; then
+                    bk_run_context --results-dir ../results --exp CASE7
                     bk_profiler "$qws_profiler_tool" --level "$qws_profiler_level" --archive ../results/padata0.tgz --raw-dir pa -- mpiexec -n 8 ./main "${qws_case7_args[@]}" > CASE7.profile
                 # else
                 #     emit_qws_dummy_padata ../results/padata0.tgz
diff --git a/programs/sbd/README.md b/programs/sbd/README.md
index 8e92c79..01eb1b0 100644
--- a/programs/sbd/README.md
+++ b/programs/sbd/README.md
@@ -17,9 +17,22 @@ and then uses that plan for rank-0 Nsight Compute acquisition when
 Use `BK_SBD_NCU_PROFILE_MODE=discovery-only` to stop after the NSYS discovery
 and plan generation step. `BK_SBD_NCU_PLAN_TOP_K` controls how many kernels are
 selected; discovery-only defaults to all kernels, while discovery defaults to
-the top three. Generated NCU archives are attached to the `mult` section as
+the top three. The RIKYU system default overrides this selection to the top one.
+Generated NCU archives are attached to the `mult` section as
 measurement artifacts.
 
+## Workflow stage timing
+
+SBD uses the common execution helpers for its ordinary benchmark, NSYS
+collection/export, NCU plan generation and NCU collection/export. Benchkit
+automatically logs and stores their stage timings as Measurement Artifacts;
+the application does not manage timing records. See the
+[shared profiler guide](../../docs/guides/profiler-support.md#workflow-stage-timing)
+for scope, failure handling and interpretation. The internal Davidson FOM
+remains independent of workflow wall time.
+
+## Benchmark recipes
+
 The RIKYU recipe uses the H2O cc-pVDZ FCIDUMP with the `1em7` selected alpha
 determinant file (about 628 million product determinants), one MPI rank per
 B200 GPU, 32 OpenMP threads per rank, and the rank-distributed/index-reordered/NCCL
diff --git a/programs/sbd/profile.sh b/programs/sbd/profile.sh
index 600304b..8507e6c 100644
--- a/programs/sbd/profile.sh
+++ b/programs/sbd/profile.sh
@@ -146,8 +146,8 @@ sbd_run_rank0_nsys_discovery() {
   local profile_status
 
   echo "Running SBD NSYS kernel discovery for automatic NCU plan generation" >&2
-  set +e
-  mpirun -np "$n_ranks" bash -lc '
+  profile_status=0
+  bk_profile_execute --tool nsys --phase collect --log "$log_file" -- mpirun -np "$n_ranks" bash -lc '
     rank=${OMPI_COMM_WORLD_RANK:-${PMIX_RANK:-${SLURM_PROCID:-0}}}
     local_rank=${OMPI_COMM_WORLD_LOCAL_RANK:-${SLURM_LOCALID:-0}}
     export CUDA_VISIBLE_DEVICES="${local_rank}"
@@ -157,9 +157,7 @@ sbd_run_rank0_nsys_discovery() {
       exec nsys profile --force-overwrite=true --trace=cuda --sample=none -o "$report_base" ./diag "$@"
     fi
     exec ./diag "$@"
-  ' bash "$report_base" "$@" > "$log_file" 2>&1
-  profile_status=$?
-  set -e
+  ' bash "$report_base" "$@" || profile_status=$?
 
   if [ "$profile_status" -ne 0 ]; then
     echo "SBD NSYS kernel discovery failed with status ${profile_status}" >&2
@@ -225,14 +223,12 @@ sbd_generate_ncu_plan() {
       return 1
     fi
 
-    set +e
-    nsys stats --force-export=true \
+    nsys_stats_status=0
+    bk_profile_execute --tool nsys --phase export -- nsys stats --force-export=true \
       --report cuda_gpu_kern_sum,cuda_api_sum \
       --format csv \
       --output "$nsys_stats_base" \
-      "$nsys_report" >/dev/null
-    nsys_stats_status=$?
-    set -e
+      "$nsys_report" >/dev/null || nsys_stats_status=$?
     if [ "$nsys_stats_status" -ne 0 ]; then
       echo "SBD NSYS CUDA summary export failed with status ${nsys_stats_status}" >&2
       return "$nsys_stats_status"
@@ -263,7 +259,7 @@ sbd_generate_ncu_plan() {
     esac
   fi
 
-  if ! "$python_bin" "${SBD_BENCHKIT_ROOT}/scripts/profiling/generate_ncu_plan.py" \
+  if ! bk_generate_ncu_plan \
     --nsys-csv "$discovery_csv" \
     --out-discovery "$discovery_json" \
     --out-plan "$plan_json" \
@@ -394,14 +390,14 @@ sbd_run_rank0_ncu_profile() {
     ' bash "$profile_base" "${#ncu_level_args[@]}" "${ncu_level_args[@]}" \
       "$kernel_regex" "$launch_skip" "$launch_count" "$@"
   )
-  set +e
+  profiler_status=0
   if [ "$profile_timeout_seconds" -gt 0 ]; then
-    timeout --kill-after=60s "$profile_timeout_seconds" "${profile_cmd[@]}" > "$profile_log" 2>&1
+    bk_profile_execute --tool ncu --phase collect --profile "$profile_slug" --log "$profile_log" -- \
+      timeout --kill-after=60s "$profile_timeout_seconds" "${profile_cmd[@]}" || profiler_status=$?
   else
-    "${profile_cmd[@]}" > "$profile_log" 2>&1
+    bk_profile_execute --tool ncu --phase collect --profile "$profile_slug" --log "$profile_log" -- \
+      "${profile_cmd[@]}" || profiler_status=$?
   fi
-  profiler_status=$?
-  set -e
 
   if [ "$profiler_status" -eq 0 ]; then
     echo "bk_profiler[ncu]: completed ${rep_name} level=${profiler_level}" >&2
@@ -411,12 +407,13 @@ sbd_run_rank0_ncu_profile() {
 
   report_file=$(bk_profiler_find_ncu_report "$rep_dir" || true)
   if [ -n "$report_file" ]; then
-    ncu --import "$report_file" \
+    bk_profile_execute --tool ncu --phase export --profile "${profile_slug}/raw" -- ncu --import "$report_file" \
       --page raw \
       --csv \
       --print-units base \
       --print-fp > "${rep_dir}/profile_raw.csv" 2> "${rep_dir}/profile_raw.csv.log" || true
-    ncu --import "$report_file" --page details > "$stage_dir/reports/ncu_import_${rep_name}.txt" 2>&1 || true
+    bk_profile_execute --tool ncu --phase export --profile "${profile_slug}/details" -- \
+      ncu --import "$report_file" --page details > "$stage_dir/reports/ncu_import_${rep_name}.txt" 2>&1 || true
   fi
 
   cp -R "$rep_dir" "$stage_dir/raw/${rep_name}"
diff --git a/programs/sbd/run.sh b/programs/sbd/run.sh
index 336ff89..bf90f76 100755
--- a/programs/sbd/run.sh
+++ b/programs/sbd/run.sh
@@ -70,6 +70,8 @@ case "${system}" in
     ;;
 esac
 
+bk_run_context --results-dir "${RESULTS_DIR}" --exp "${experiment}"
+
 for input_file in fcidump.txt "${determinant_file}"; do
   if [[ ! -f "${INPUT_DIR}/${input_file}" ]]; then
     echo "SBD input not found: ${INPUT_DIR}/${input_file}" >&2
@@ -105,12 +107,12 @@ diag_args=(
 )
 
 if [[ "${system}" == "RC_FX700" ]]; then
-  mpirun -np "${n_ranks}" -bind-to numa ./diag "${diag_args[@]}" \
-    > diag.log 2>&1
+  bk_run --log diag.log -- \
+    mpirun -np "${n_ranks}" -bind-to numa ./diag "${diag_args[@]}"
 else
-  mpirun -np "${n_ranks}" bash -lc \
+  bk_run --log diag.log -- mpirun -np "${n_ranks}" bash -lc \
     'export CUDA_VISIBLE_DEVICES=$OMPI_COMM_WORLD_LOCAL_RANK; exec "$@"' \
-    bash ./diag "${diag_args[@]}" > diag.log 2>&1
+    bash ./diag "${diag_args[@]}"
 fi
 
 davidson_time=$(grep -E 'Elapsed time for davidson ' diag.log | tail -n 1 | awk '{print $(NF-1)}')
diff --git a/result_server/tests/test_result_detail_template.py b/result_server/tests/test_result_detail_template.py
index 57baf9a..f64c093 100644
--- a/result_server/tests/test_result_detail_template.py
+++ b/result_server/tests/test_result_detail_template.py
@@ -397,6 +397,74 @@ def test_public_surface_keeps_padata_archive_links(self, app):
         assert "Measurement Artifacts" in html
         assert f'href="/results/{filename}"' in html
 
+    @pytest.mark.parametrize("collection", ["sections", "overlaps"])
+    @pytest.mark.parametrize("uploaded", [True, False])
+    @pytest.mark.parametrize("public_surface", [True, False])
+    def test_profile_metadata_links_respect_surface_and_upload_state(
+        self, app, collection, uploaded, public_surface,
+    ):
+        basenames = ["kernel_discovery.json", "ncu_plan.json", "padata_kernel.metadata.json"]
+        filenames = [
+            "measurement_artifact_20250101_120000_"
+            f"12345678-1234-1234-1234-123456789abc_{name}"
+            for name in basenames
+        ]
+        result = {
+            "code": "demoapp",
+            "_server_uuid": "12345678-1234-1234-1234-123456789abc",
+            "_server_timestamp": "20250101_120000",
+            "fom_breakdown": {
+                collection: [{
+                    "name": "kernel",
+                    "artifacts": [
+                        {"type": "file_reference", "path": f"results/{name}"}
+                        for name in basenames
+                    ],
+                }],
+            },
+        }
+
+        with app.test_request_context():
+            html = _render_result_detail(
+                result, {}, filenames if uploaded else [], public_surface=public_surface,
+            )
+
+        for basename, filename in zip(basenames, filenames):
+            if public_surface:
+                assert basename not in html
+            else:
+                assert "Profile metadata" in html
+                assert f"results/{basename}" in html
+                if uploaded:
+                    assert f'href="/results/{filename}"' in html
+                else:
+                    assert f"{filename} not uploaded" in html
+                    assert f'href="/results/{filename}"' not in html
+
+    @pytest.mark.parametrize("artifact_path", [
+        "../metadata.json",
+        "results/../metadata.json",
+        "/results/metadata.json",
+        "artifacts/metadata.json",
+        "results/bad name.json",
+        "results/metadata.txt",
+    ])
+    def test_invalid_profile_metadata_paths_are_omitted(self, app, artifact_path):
+        result = {
+            "_server_uuid": "12345678-1234-1234-1234-123456789abc",
+            "_server_timestamp": "20250101_120000",
+            "fom_breakdown": {
+                "sections": [{
+                    "name": "kernel",
+                    "artifacts": [{"type": "file_reference", "path": artifact_path}],
+                }],
+            },
+        }
+        with app.test_request_context():
+            context = build_result_detail_context(result, {})
+
+        assert context["measurement_artifact_rows"] == []
+
     def test_timing_observation_artifact_is_linked_on_console_surface(self, app):
         result = {
             **FULL_RESULT,
diff --git a/result_server/utils/result_detail_view.py b/result_server/utils/result_detail_view.py
index 8142204..0eabfd6 100644
--- a/result_server/utils/result_detail_view.py
+++ b/result_server/utils/result_detail_view.py
@@ -113,7 +113,7 @@ def build_result_detail_context(
         "measurement_artifact_rows": _build_measurement_artifact_rows(
             result,
             measurement_artifact_filenames or [],
-            include_timing=not public_surface,
+            include_private=not public_surface,
         ),
         "node_status_rows": (
             [] if public_surface else _build_node_status_rows(result.get("node_status_snapshot"))
@@ -220,7 +220,7 @@ def _build_measurement_artifact_rows(
     result,
     measurement_artifact_filenames,
     *,
-    include_timing=True,
+    include_private=True,
 ):
     result_uuid = result.get("_server_uuid")
     timestamp = result.get("_server_timestamp")
@@ -228,8 +228,10 @@ def _build_measurement_artifact_rows(
         return []
 
     uploaded = set(measurement_artifact_filenames)
-    rows = _build_profile_measurement_artifact_rows(result, timestamp, result_uuid, uploaded)
-    if include_timing:
+    rows = _build_profile_measurement_artifact_rows(
+        result, timestamp, result_uuid, uploaded, include_metadata=include_private,
+    )
+    if include_private:
         rows.extend(
             _build_timing_measurement_artifact_rows(result, timestamp, result_uuid, uploaded)
         )
@@ -241,7 +243,9 @@ def _build_measurement_artifact_rows(
     return rows
 
 
-def _build_profile_measurement_artifact_rows(result, timestamp, result_uuid, uploaded):
+def _build_profile_measurement_artifact_rows(
+    result, timestamp, result_uuid, uploaded, *, include_metadata=True,
+):
     rows = []
     breakdown = result.get("fom_breakdown")
     if not isinstance(breakdown, dict):
@@ -261,11 +265,19 @@ def _build_profile_measurement_artifact_rows(result, timestamp, result_uuid, upl
                     result_uuid,
                     artifact_path,
                 )
+                kind = "Profile archive"
+                if not candidates and include_metadata:
+                    filename = stored_measurement_artifact_filename_from_path(
+                        timestamp, result_uuid, artifact_path,
+                    )
+                    if filename.endswith(".json"):
+                        candidates = [filename]
+                        kind = "Profile metadata"
                 if not candidates:
                     continue
                 filename = _choose_uploaded_filename(candidates, uploaded)
                 rows.append({
-                    "kind": "Profile archive",
+                    "kind": kind,
                     "source": f"{source_label}: {item_name}",
                     "artifact_path": artifact_path,
                     "filename": filename,
diff --git a/scripts/bk_functions.sh b/scripts/bk_functions.sh
index d1ab81b..4c084dd 100644
--- a/scripts/bk_functions.sh
+++ b/scripts/bk_functions.sh
@@ -9,6 +9,110 @@ if [ -z "${BK_BENCHKIT_ROOT:-}" ]; then
   export BK_BENCHKIT_ROOT
 fi
 
+if [ -z "${_BK_WORKFLOW_SESSION_ID:-}" ]; then
+  printf -v _BK_WORKFLOW_EPOCH '%(%s)T' -1
+  export _BK_WORKFLOW_SESSION_ID="${_BK_WORKFLOW_EPOCH}-${BASHPID}-${RANDOM}"
+fi
+
+# Set execution scope once, before changing directory or entering a pipeline.
+bk_run_context() {
+  local results_dir="${BK_RUN_RESULTS_DIR:-${PWD}/results}"
+  local run_exp="${BK_RUN_EXP:-}"
+  while [ "$#" -gt 0 ]; do
+    case "$1" in
+      --results-dir|--exp)
+        [ "$#" -ge 2 ] || return 2
+        case "$1" in
+          --results-dir) results_dir="$2" ;;
+          --exp) run_exp="$2" ;;
+        esac
+        shift 2
+        ;;
+      *) echo "bk_run_context: unknown argument" >&2; return 2 ;;
+    esac
+  done
+  case "$results_dir" in
+    /*) ;;
+    *) results_dir="${PWD}/${results_dir}" ;;
+  esac
+  export BK_RUN_RESULTS_DIR="$results_dir" BK_RUN_EXP="$run_exp"
+  # An unavailable recorder must not leave a previous run eligible for collection.
+  if ! rm -f "${results_dir}/.workflow_session.json"; then
+    echo "Benchkit timing: previous execution context could not be invalidated" >&2
+    return 0
+  fi
+  "${PYTHON_BIN:-python3}" "${BK_BENCHKIT_ROOT}/scripts/profiling/workflow_timing.py" \
+    --results-dir "$results_dir" context --session "$_BK_WORKFLOW_SESSION_ID" || \
+    echo "Benchkit timing: execution context could not be recorded" >&2
+  return 0
+}
+
+_bk_execute_command() {
+  local stage="$1" tool="$2" profile="$3" log_file="$4"
+  shift 4
+  local token="" command_status=0
+  local results_dir="${BK_RUN_RESULTS_DIR:-${BK_BENCHKIT_ROOT}/results}"
+  local recorder="${BK_BENCHKIT_ROOT}/scripts/profiling/workflow_timing.py"
+  token=$("${PYTHON_BIN:-python3}" "$recorder" --results-dir "$results_dir" \
+    start --session "$_BK_WORKFLOW_SESSION_ID" --exp "${BK_RUN_EXP:-}" \
+    --stage "$stage" --tool "$tool" --profile "$profile") || token=""
+  if [ -n "$log_file" ]; then
+    "$@" > "$log_file" 2>&1 || command_status=$?
+  else
+    "$@" || command_status=$?
+  fi
+  if [ -n "$token" ]; then
+    "${PYTHON_BIN:-python3}" "$recorder" --results-dir "$results_dir" \
+      finish "$token" "$command_status" || \
+      echo "Benchkit timing: stage completion could not be recorded" >&2
+  fi
+  return "$command_status"
+}
+
+bk_run() {
+  local log_file=""
+  if [ "${1:-}" = --log ]; then
+    [ "$#" -ge 2 ] || return 2
+    log_file="$2"
+    shift 2
+  fi
+  [ "${1:-}" = -- ] && shift
+  [ "$#" -gt 0 ] || return 2
+  _bk_execute_command benchmark none "" "$log_file" "$@"
+}
+
+# Execute a prepared profiler command, including MPI/container launchers.
+bk_profile_execute() {
+  local tool="" phase="collect" profile="" log_file=""
+  while [ "$#" -gt 0 ]; do
+    case "$1" in
+      --tool|--phase|--profile|--log)
+        [ "$#" -ge 2 ] || return 2
+        case "$1" in
+          --tool) tool="$2" ;;
+          --phase) phase="$2" ;;
+          --profile) profile="$2" ;;
+          --log) log_file="$2" ;;
+        esac
+        shift 2
+        ;;
+      --) shift; break ;;
+      *) echo "bk_profile_execute: unknown argument" >&2; return 2 ;;
+    esac
+  done
+  [ -n "$tool" ] && [ "$#" -gt 0 ] || return 2
+  case "$phase" in
+    collect|export|plan) ;;
+    *) echo "bk_profile_execute: unknown phase" >&2; return 2 ;;
+  esac
+  _bk_execute_command "$phase" "$tool" "$profile" "$log_file" "$@"
+}
+
+bk_generate_ncu_plan() {
+  bk_profile_execute --tool ncu --phase plan -- \
+    "${PYTHON_BIN:-python3}" "${BK_BENCHKIT_ROOT}/scripts/profiling/generate_ncu_plan.py" "$@"
+}
+
 # bk_emit_result - Output a standardized FOM result line.
 #
 # Named arguments:
@@ -2313,6 +2417,8 @@ bk_run_ncu_acquisition_profile() {
 }
 
 bk_profiler() {
+  local BK_RUN_RESULTS_DIR="${BK_RUN_RESULTS_DIR:-}"
+  local _bk_workflow_profile
   if [ $# -lt 2 ]; then
     echo "bk_profiler: requires a profiler tool and an execution command" >&2
     return 1
@@ -2375,11 +2481,14 @@ bk_profiler() {
     return 1
   fi
 
+  BK_RUN_RESULTS_DIR="${BK_RUN_RESULTS_DIR:-$(dirname "$_bk_profiler_archive")}"
   if [ -z "$_bk_profiler_tool" ]; then
-    "$@"
+    bk_run -- "$@"
     return $?
   fi
 
+  _bk_workflow_profile=$(basename "$_bk_profiler_archive" .tgz)
+
   _bk_profiler_level=$(bk_get_profiler_level "$_bk_profiler_tool" "$_bk_profiler_level") || return 1
   _bk_profiler_report_format=$(bk_get_profiler_report_format "$_bk_profiler_tool" "$_bk_profiler_level" "$_bk_profiler_report_format") || return 1
 
@@ -2410,7 +2519,8 @@ bk_profiler() {
         bk_profiler_call_optional_hook bk_profiler_before_run "$_bk_profiler_tool" "$_bk_profiler_level" "$_bk_fapp_rep_name" "$_bk_fapp_event" "$@" || return 1
         # BK_PROFILER_ARGS is intentionally word-split into fapp options.
         # shellcheck disable=SC2086
-        if fapp -C -d "$_bk_fapp_rep_dir" ${_bk_profiler_extra_args} -Hevent="${_bk_fapp_event}" "$@"; then
+        if bk_profile_execute --tool fapp --profile "${_bk_workflow_profile}/${_bk_fapp_rep_name}/${_bk_fapp_event}" -- \
+          fapp -C -d "$_bk_fapp_rep_dir" ${_bk_profiler_extra_args} -Hevent="${_bk_fapp_event}" "$@"; then
           _bk_fapp_status=0
         else
           _bk_fapp_status=$?
@@ -2450,7 +2560,8 @@ bk_profiler() {
       bk_profiler_call_optional_hook bk_profiler_before_run "$_bk_profiler_tool" "$_bk_profiler_level" "$_bk_ncu_rep_name" "$_bk_profiler_level" "$@" || return 1
       # BK_PROFILER_ARGS is intentionally word-split into ncu options.
       # shellcheck disable=SC2086
-      if ncu -o "$_bk_ncu_profile_base" --target-processes all ${_bk_ncu_level_args} ${_bk_profiler_extra_args} "$@"; then
+      if bk_profile_execute --tool ncu --profile "${_bk_workflow_profile}/${_bk_ncu_rep_name}" -- \
+        ncu -o "$_bk_ncu_profile_base" --target-processes all ${_bk_ncu_level_args} ${_bk_profiler_extra_args} "$@"; then
         _bk_profiler_status=0
       else
         _bk_profiler_status=$?
@@ -2465,7 +2576,8 @@ bk_profiler() {
         1|true|TRUE|yes|YES|on|ON)
           _bk_ncu_report_file=$(bk_profiler_find_ncu_report "$_bk_ncu_rep_dir" || true)
           if [ -n "$_bk_ncu_report_file" ]; then
-            ncu --import "$_bk_ncu_report_file" \
+            bk_profile_execute --tool ncu --phase export --profile "${_bk_workflow_profile}/${_bk_ncu_rep_name}/raw" -- \
+              ncu --import "$_bk_ncu_report_file" \
               --page raw \
               --csv \
               --print-units base \
@@ -2501,12 +2613,14 @@ bk_profiler() {
           if [ "$_bk_profiler_report_format" = "text" ] || [ "$_bk_profiler_report_format" = "both" ]; then
             # BK_PROFILER_REPORT_ARGS is intentionally word-split into fapp/fapppx options.
             # shellcheck disable=SC2086
-            "$_bk_fapp_post_cmd" -A -d "$_bk_fapp_rep_dir" ${_bk_profiler_report_extra_args} > "$_bk_stage_dir/reports/fapp_A_${_bk_fapp_rep_name}.txt" 2>&1 || true
+            bk_profile_execute --tool fapp --phase export --profile "${_bk_workflow_profile}/${_bk_fapp_rep_name}/text" -- \
+              "$_bk_fapp_post_cmd" -A -d "$_bk_fapp_rep_dir" ${_bk_profiler_report_extra_args} > "$_bk_stage_dir/reports/fapp_A_${_bk_fapp_rep_name}.txt" 2>&1 || true
           fi
           if [ "$_bk_profiler_report_format" = "csv" ] || [ "$_bk_profiler_report_format" = "both" ]; then
             # BK_PROFILER_REPORT_ARGS is intentionally word-split into fapp/fapppx options.
             # shellcheck disable=SC2086
-            "$_bk_fapp_post_cmd" -A -d "$_bk_fapp_rep_dir" ${_bk_profiler_report_extra_args} -Icpupa -tcsv -o "$_bk_stage_dir/reports/cpu_pa_${_bk_fapp_rep_name}.csv" >/dev/null 2>&1 || true
+            bk_profile_execute --tool fapp --phase export --profile "${_bk_workflow_profile}/${_bk_fapp_rep_name}/csv" -- \
+              "$_bk_fapp_post_cmd" -A -d "$_bk_fapp_rep_dir" ${_bk_profiler_report_extra_args} -Icpupa -tcsv -o "$_bk_stage_dir/reports/cpu_pa_${_bk_fapp_rep_name}.csv" >/dev/null 2>&1 || true
           fi
         done
       else
@@ -2520,7 +2634,8 @@ bk_profiler() {
         if [ -n "$_bk_ncu_report_file" ] && { [ "$_bk_profiler_report_format" = "text" ] || [ "$_bk_profiler_report_format" = "both" ]; }; then
           # BK_PROFILER_REPORT_ARGS is intentionally word-split into ncu --import options.
           # shellcheck disable=SC2086
-          ncu --import "$_bk_ncu_report_file" --page details ${_bk_profiler_report_extra_args} > "$_bk_stage_dir/reports/ncu_import_${_bk_ncu_rep_name}.txt" 2>&1 || true
+          bk_profile_execute --tool ncu --phase export --profile "${_bk_workflow_profile}/${_bk_ncu_rep_name}/details" -- \
+            ncu --import "$_bk_ncu_report_file" --page details ${_bk_profiler_report_extra_args} > "$_bk_stage_dir/reports/ncu_import_${_bk_ncu_rep_name}.txt" 2>&1 || true
         fi
       done
       ;;
diff --git a/scripts/profiling/workflow_timing.py b/scripts/profiling/workflow_timing.py
new file mode 100644
index 0000000..db4c58f
--- /dev/null
+++ b/scripts/profiling/workflow_timing.py
@@ -0,0 +1,179 @@
+"""Persist common execution stages and expose scoped timing observations."""
+
+import argparse
+from datetime import datetime, timezone
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sys
+import tempfile
+import time
+import uuid
+
+
+def timestamp():
+    return datetime.now(timezone.utc).isoformat(timespec="milliseconds")
+
+
+def write_document(path, document):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(
+            mode="w", encoding="utf-8", dir=path.parent, delete=False
+        ) as handle:
+            temporary = handle.name
+            json.dump(document, handle, indent=2)
+            handle.write("\n")
+        os.replace(temporary, path)
+    finally:
+        if temporary and os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--results-dir", type=Path, required=True)
+    commands = parser.add_subparsers(dest="command", required=True)
+    start = commands.add_parser("start")
+    start.add_argument("--session", required=True)
+    start.add_argument("--exp", default="")
+    start.add_argument("--stage", required=True)
+    start.add_argument("--tool", default="none")
+    start.add_argument("--profile", default="")
+    finish = commands.add_parser("finish")
+    finish.add_argument("token")
+    finish.add_argument("exit_code", type=int)
+    context = commands.add_parser("context")
+    context.add_argument("--session", required=True)
+    commands.add_parser("manifest")
+    args = parser.parse_args()
+
+    if args.command == "manifest" and not args.results_dir.exists():
+        print(json.dumps({"schema_version": 1, "observations": []}))
+        return
+    args.results_dir.mkdir(parents=True, exist_ok=True)
+    # A scope may be entered from command substitution, pipelines or concurrent workers.
+    with (args.results_dir / ".workflow_timing.lock").open("a") as lock:
+        fcntl.flock(lock, fcntl.LOCK_EX)
+        if args.command == "manifest":
+            print(json.dumps(collect_manifest(args.results_dir)))
+            return
+        if args.command == "context":
+            write_document(args.results_dir / ".workflow_session.json", {"session_id": args.session})
+            return
+        update_stage(args)
+
+
+def update_stage(args):
+    if args.command == "start":
+        scope = hashlib.sha256(args.exp.encode("utf-8")).hexdigest()
+        path = args.results_dir / f"workflow_timing_{scope}.json"
+        document = {}
+        if path.exists():
+            with path.open(encoding="utf-8") as handle:
+                document = json.load(handle)
+        if document.get("session_id") != args.session:
+            document = {
+                "schema_version": 1,
+                "kind": "workflow_stage_timing",
+                "producer": "benchkit",
+                "session_id": args.session,
+                "exp": args.exp,
+                "elapsed_clock": "monotonic",
+                "stages": [],
+            }
+        record = {
+            "id": uuid.uuid4().hex,
+            "stage": args.stage,
+            "tool": args.tool,
+            "profile": args.profile,
+            "started_at": timestamp(),
+            "started_monotonic_ns": time.monotonic_ns(),
+            "status": "running",
+        }
+        document["stages"].append(record)
+    else:
+        filename, record_id = args.token.split(":", 1)
+        if not re.fullmatch(r"workflow_timing_[0-9a-f]{64}\.json", filename):
+            raise ValueError("invalid stage token")
+        path = args.results_dir / filename
+        with path.open(encoding="utf-8") as handle:
+            document = json.load(handle)
+        record = next(item for item in document["stages"] if item["id"] == record_id)
+        if record["status"] != "running":
+            raise ValueError("stage has already finished")
+        elapsed = (time.monotonic_ns() - record.pop("started_monotonic_ns")) / 1e9
+        record.update({
+            "finished_at": timestamp(),
+            "elapsed_seconds": elapsed,
+            "exit_code": args.exit_code,
+            "status": "completed" if args.exit_code == 0 else "failed",
+        })
+    write_document(path, document)
+    if args.command == "start":
+        write_document(args.results_dir / ".workflow_session.json", {"session_id": args.session})
+    label = record["stage"]
+    label += " tool=" + record["tool"]
+    if record["profile"]:
+        label += " profile=" + record["profile"]
+    if args.command == "start":
+        print(f"Benchkit timing: stage={label} started_at={record['started_at']}", file=sys.stderr)
+        print(f"{path.name}:{record['id']}")
+    else:
+        print(
+            f"Benchkit timing: stage={label} finished_at={record['finished_at']} "
+            f"elapsed_seconds={elapsed:.3f} status={record['status']} "
+            f"exit_code={args.exit_code}", file=sys.stderr,
+        )
+
+
+def collect_manifest(results_dir):
+    manifest = {"schema_version": 1, "observations": []}
+    session_path = results_dir / ".workflow_session.json"
+    if not session_path.exists():
+        return manifest
+    with session_path.open(encoding="utf-8") as handle:
+        session = json.load(handle)["session_id"]
+    for path in sorted(results_dir.glob("workflow_timing_*.json")):
+        if path.is_symlink() or not re.fullmatch(r"workflow_timing_[0-9a-f]{64}\.json", path.name):
+            continue
+        try:
+            with path.open(encoding="utf-8") as handle:
+                document = json.load(handle)
+            if document.get("session_id") != session:
+                continue
+            if document.get("kind") != "workflow_stage_timing" or document.get("schema_version") != 1:
+                continue
+            stages = document["stages"]
+            statuses = [stage["status"] for stage in stages]
+            observation = {
+                "id": path.stem,
+                "kind": "workflow-stage-timing",
+                "producer": "benchkit",
+                "format": "workflow_stage_timing/v1",
+                "artifact": {"type": "file_reference", "path": f"results/{path.name}"},
+                "summary": {
+                    "stage_count": len(stages),
+                    "completed_count": statuses.count("completed"),
+                    "failed_count": statuses.count("failed"),
+                    "unfinished_count": statuses.count("running"),
+                },
+            }
+            if document.get("exp"):
+                observation["result_exp"] = document["exp"]
+            manifest["observations"].append(observation)
+        except (OSError, ValueError, KeyError, TypeError, AttributeError):
+            print("Benchkit timing: skipped invalid stage artifact", file=sys.stderr)
+    return manifest
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (OSError, ValueError, KeyError, StopIteration, TypeError, AttributeError):
+        print("Benchkit timing: unable to access stage artifact", file=sys.stderr)
+        sys.exit(1)
diff --git a/scripts/result.sh b/scripts/result.sh
index b481fcf..349b2bf 100644
--- a/scripts/result.sh
+++ b/scripts/result.sh
@@ -803,9 +803,20 @@ fi
 
 build_timing_observations_block() {
   local timing_observations_file="results/timing_observations.json"
+  local workflow_json='{"schema_version":1,"observations":[]}'
+  local workflow_recorder
+  workflow_recorder="$(dirname "${BASH_SOURCE[0]}")/profiling/workflow_timing.py"
+  if [ -f results/.workflow_session.json ]; then
+    if ! workflow_json=$("${PYTHON_BIN:-python3}" "$workflow_recorder" --results-dir results manifest); then
+      echo "WARNING: workflow timing observations could not be collected" >&2
+      workflow_json='{"schema_version":1,"observations":[]}'
+    fi
+  fi
 
   if [ ! -f "$timing_observations_file" ]; then
-    printf '%s' ""
+    if printf '%s' "$workflow_json" | jq -e '.observations | length > 0' >/dev/null; then
+      printf '%s' "$workflow_json"
+    fi
     return 0
   fi
 
@@ -839,7 +850,8 @@ build_timing_observations_block() {
     return 1
   fi
 
-  printf '%s' "$timing_observations_json"
+  printf '%s' "$timing_observations_json" | jq -cS --argjson workflow "$workflow_json" \
+    '.observations += $workflow.observations'
 }
 
 if ! timing_observations_block=$(build_timing_observations_block); then
diff --git a/scripts/tests/run_profile_data_shell_tests.sh b/scripts/tests/run_profile_data_shell_tests.sh
index 67f4c1c..bbed0da 100644
--- a/scripts/tests/run_profile_data_shell_tests.sh
+++ b/scripts/tests/run_profile_data_shell_tests.sh
@@ -40,6 +40,7 @@ tests_for_group() {
     common)
       cat <<'EOF'
 scripts/tests/test_bk_profiler.sh
+scripts/tests/test_workflow_timing.sh
 scripts/tests/test_bk_fetch_source.sh
 scripts/tests/test_bk_input_info.sh
 scripts/tests/test_bk_timing_observations.sh
diff --git a/scripts/tests/test_bk_profiler.sh b/scripts/tests/test_bk_profiler.sh
index 13dbf24..dfce16f 100644
--- a/scripts/tests/test_bk_profiler.sh
+++ b/scripts/tests/test_bk_profiler.sh
@@ -8,6 +8,15 @@ source "${REPO_DIR}/scripts/bk_functions.sh"
 
 TMP_DIR=$(mktemp -d)
 trap 'rm -rf "${TMP_DIR}"' EXIT
+TMP_RESULTS="${TMP_DIR}/results"
+mkdir -p "$TMP_RESULTS"
+bk_run_context --results-dir "$TMP_RESULTS" --exp profiler
+# Keep profiler staging and implicit timing artifacts out of the source tree.
+cd "$TMP_DIR"
+
+timing_stages() {
+  jq -s '[.[].stages[]]' "$TMP_RESULTS"/workflow_timing_*.json
+}
 
 test "$(bk_resolve_profiler_tool fapp)" = "fapp"
 (export BK_PROFILER=ncu; test "$(bk_resolve_profiler_tool fapp)" = "ncu")
@@ -64,6 +73,9 @@ EOF
 cat > "${FAKE_BIN}/fapppx" <<'EOF'
 #!/bin/bash
 set -euo pipefail
+if [ "${FAKE_EXPORT_FAIL:-0}" = 1 ]; then
+  exit 17
+fi
 dir=""
 outfile=""
 while [ $# -gt 0 ]; do
@@ -124,6 +136,9 @@ while [ $# -gt 0 ]; do
 done
 
 if [ "$import_mode" -eq 1 ]; then
+  if [ "${FAKE_EXPORT_FAIL:-0}" = 1 ]; then
+    exit 17
+  fi
   printf 'ncu import:%s\n' "$import_file"
   exit 0
 fi
@@ -147,6 +162,10 @@ run_and_check_level() {
   local archive="${TMP_DIR}/${level}.tgz"
   local extract_dir="${TMP_DIR}/${level}_extract"
   local raw_dir="${TMP_DIR}/${level}_pa"
+  local before_count=0
+  if compgen -G "$TMP_RESULTS/workflow_timing_*.json" >/dev/null; then
+    before_count=$(timing_stages | jq 'length')
+  fi
 
   bk_profiler fapp --level "$level" --archive "$archive" --raw-dir "$raw_dir" -- true
   mkdir -p "$extract_dir"
@@ -166,6 +185,17 @@ run_and_check_level() {
   grep -q "\"level\": \"${level}\"" "${extract_dir}/bk_profiler_artifact/meta.json"
   grep -q "\"report_format\": \"${expected_format}\"" "${extract_dir}/bk_profiler_artifact/meta.json"
   grep -q "\"event\": \"${expected_last_event}\"" "${extract_dir}/bk_profiler_artifact/meta.json"
+  local exports_per_rep=1
+  if [ "$expect_csv" = yes ]; then
+    exports_per_rep=2
+  fi
+  timing_stages | jq -e --argjson start "$before_count" \
+    --argjson reps "$expected_last_rep" --argjson exports "$exports_per_rep" '
+    .[$start:] |
+    ([.[] | select(.stage == "collect")] | length) == $reps and
+    ([.[] | select(.stage == "export")] | length) == ($reps * $exports) and
+    all(.[]; .tool == "fapp" and .status == "completed" and .exit_code == 0)
+  ' >/dev/null
 }
 
 run_and_check_level single 1 pa1 text no
@@ -176,7 +206,12 @@ run_and_check_level detailed 17 pa17 both yes
 ncu_archive="${TMP_DIR}/ncu.tgz"
 ncu_extract="${TMP_DIR}/ncu_extract"
 ncu_raw="${TMP_DIR}/ncu_pa"
+before_ncu=$(timing_stages | jq 'length')
 bk_profiler ncu --level single --archive "$ncu_archive" --raw-dir "$ncu_raw" -- bash -c 'printf "ncu target\n"'
+timing_stages | jq -e --argjson start "$before_ncu" '
+  .[$start:] | map([.stage, .tool, .status, .exit_code]) ==
+  [["collect", "ncu", "completed", 0], ["export", "ncu", "completed", 0]]
+' >/dev/null
 mkdir -p "$ncu_extract"
 tar -xzf "$ncu_archive" -C "$ncu_extract"
 test -f "${ncu_extract}/bk_profiler_artifact/meta.json"
@@ -209,7 +244,13 @@ ncu_raw_csv_archive="${TMP_DIR}/ncu_raw_csv.tgz"
 ncu_raw_csv_extract="${TMP_DIR}/ncu_raw_csv_extract"
 ncu_raw_csv_raw="${TMP_DIR}/ncu_raw_csv_pa"
 export BK_PROFILER_NCU_RAW_CSV=true
+before_ncu=$(timing_stages | jq 'length')
 bk_profiler ncu --level single --archive "$ncu_raw_csv_archive" --raw-dir "$ncu_raw_csv_raw" -- bash -c 'printf "ncu raw csv target\n"'
+timing_stages | jq -e --argjson start "$before_ncu" '
+  .[$start:] | map([.stage, .tool, .status, .exit_code]) ==
+  [["collect", "ncu", "completed", 0], ["export", "ncu", "completed", 0],
+   ["export", "ncu", "completed", 0]]
+' >/dev/null
 unset BK_PROFILER_NCU_RAW_CSV
 mkdir -p "$ncu_raw_csv_extract"
 tar -xzf "$ncu_raw_csv_archive" -C "$ncu_raw_csv_extract"
@@ -302,6 +343,8 @@ else
 fi
 unset FAKE_FAPP_FAIL
 test "$fapp_fail_status" -eq 23
+timing_stages | jq -e 'any(.[]; .stage == "collect" and .tool == "fapp" and
+  .status == "failed" and .exit_code == 23)' >/dev/null
 mkdir -p "$fapp_fail_extract"
 tar -xzf "$fapp_fail_archive" -C "$fapp_fail_extract"
 test -f "${fapp_fail_extract}/bk_profiler_artifact/meta.json"
@@ -318,9 +361,32 @@ else
   ncu_fail_status=$?
 fi
 test "$ncu_fail_status" -eq 42
+timing_stages | jq -e 'any(.[]; .stage == "collect" and .tool == "ncu" and
+  .status == "failed" and .exit_code == 42)' >/dev/null
 mkdir -p "$ncu_fail_extract"
 tar -xzf "$ncu_fail_archive" -C "$ncu_fail_extract"
 test -f "${ncu_fail_extract}/bk_profiler_artifact/meta.json"
 ! test -f "${ncu_fail_extract}/bk_profiler_artifact/raw/rep1/profile.ncu-rep"
 
+# Best-effort report failure must be visible without changing collection status.
+for tool in fapp ncu; do
+  before_export=$(timing_stages | jq 'length')
+  FAKE_EXPORT_FAIL=1 bk_profiler "$tool" --level single \
+    --archive "${TMP_DIR}/${tool}_export_fail.tgz" \
+    --raw-dir "${TMP_DIR}/${tool}_export_fail_pa" -- true
+  timing_stages | jq -e --argjson start "$before_export" --arg tool "$tool" '
+    .[$start:] | length == 2 and
+    all(.[]; .tool == $tool) and
+    .[0].stage == "collect" and .[0].status == "completed" and .[0].exit_code == 0 and
+    .[1].stage == "export" and .[1].status == "failed" and .[1].exit_code == 17
+  ' >/dev/null
+  test -f "${TMP_DIR}/${tool}_export_fail.tgz"
+done
+
+timing_stages | jq -e '
+  ([.[].id] | unique | length) == length and
+  all(.[]; (.elapsed_seconds | type == "number" and . >= 0))
+' >/dev/null
+test ! -e "$TMP_RESULTS/timing_observations.json"
+
 echo "bk_profiler tests passed"
diff --git a/scripts/tests/test_sbd_ncu_profile.sh b/scripts/tests/test_sbd_ncu_profile.sh
index 6353d8c..93ee01b 100644
--- a/scripts/tests/test_sbd_ncu_profile.sh
+++ b/scripts/tests/test_sbd_ncu_profile.sh
@@ -198,6 +198,7 @@ export BK_PROFILER_LEVEL=single
 sbd_configure_ncu_profile_from_run_env RIKYU
 test "${BK_SBD_NCU_PROFILE}" = "true"
 test "${BK_SBD_NCU_PROFILER_LEVEL}" = "single"
+bk_run_context --results-dir "${RESULTS_DIR}" --exp timing-test
 sbd_run_configured_ncu_profiles RIKYU 4 \
   --fcidump fcidump.txt \
   --adetfile h2o-1em7-alpha.txt
@@ -247,4 +248,27 @@ done
 grep -Fq 'regex:.*sbd::MultUnified<double,[[:space:]]*\(int\)0,[[:space:]]*\(int\)1>.*' "$FAKE_NCU_LOG"
 awk -F '\t' '$2 != "launch_count=1" || $3 != "launch_count_seen=1" { bad = 1 } END { exit bad }' "$FAKE_NCU_LOG"
 
+# Keep only application integration evidence here; recorder semantics are common.
+mapfile -t timing_files < <(find "${RESULTS_DIR}" -maxdepth 1 -name 'workflow_timing_*.json' | sort)
+test "${#timing_files[@]}" -eq 1
+jq -e '
+  .kind == "workflow_stage_timing" and .producer == "benchkit" and
+  .exp == "timing-test" and
+  any(.stages[]; .stage == "collect" and .tool == "nsys") and
+  any(.stages[]; .stage == "export" and .tool == "nsys") and
+  any(.stages[]; .stage == "plan" and .tool == "ncu") and
+  ([.stages[] | select(.stage == "collect" and .tool == "ncu")] | length) == 5 and
+  any(.stages[]; .stage == "export" and .tool == "ncu") and
+  all(.stages[]; .status == "completed" and .exit_code == 0)
+' "${timing_files[0]}" >/dev/null
+
+while IFS= read -r profile_name; do
+  profile_slug=$(bk_profile_slug "$profile_name")
+  jq -e --arg profile "$profile_slug" '
+    any(.stages[]; .stage == "collect" and .tool == "ncu" and .profile == $profile)
+  ' "${timing_files[0]}" >/dev/null
+done < <(jq -r '.profiles[].name' "${RESULTS_DIR}/sbd_ncu_plan.json")
+
+test ! -e "${RESULTS_DIR}/sbd_stage_timing.json"
+test ! -e "${RESULTS_DIR}/timing_observations.json"
 echo "SBD NCU profile test passed"
diff --git a/scripts/tests/test_workflow_timing.sh b/scripts/tests/test_workflow_timing.sh
new file mode 100644
index 0000000..5df471e
--- /dev/null
+++ b/scripts/tests/test_workflow_timing.sh
@@ -0,0 +1,428 @@
+#!/bin/bash
+set -euo pipefail
+
+SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+REPO_DIR=$(cd "${SCRIPT_DIR}/../.." && pwd)
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+TMP_DIR=$(mktemp -d)
+trap 'rm -rf "${TMP_DIR}"' EXIT
+trap 'echo "workflow timing test failed at line $LINENO" >&2' ERR
+
+source "${REPO_DIR}/scripts/bk_functions.sh"
+RECORDER="${REPO_DIR}/scripts/profiling/workflow_timing.py"
+RESULTS_DIR="${TMP_DIR}/results"
+export RESULTS_DIR
+mkdir -p "$RESULTS_DIR" "${TMP_DIR}/nested/work" "${TMP_DIR}/bin"
+
+manifest() {
+  "$PYTHON_BIN" "$RECORDER" --results-dir "$RESULTS_DIR" manifest
+}
+
+timing_file() {
+  local relative
+  relative=$(manifest | jq -er --arg exp "$1" \
+    '.observations[] | select((.result_exp // "") == $exp) | .artifact.path')
+  printf '%s/%s\n' "$RESULTS_DIR" "${relative#results/}"
+}
+
+# Context declaration must be lazy, including when the directory already exists.
+bk_run_context --results-dir "$RESULTS_DIR" --exp CASE0
+test -z "$(find "$RESULTS_DIR" -name 'workflow_timing_*.json' -print -quit)"
+manifest | jq -e '.schema_version == 1 and .observations == []' >/dev/null
+
+probe_command() {
+  local file
+  file=$(timing_file CASE0)
+  jq -e '
+    .schema_version == 1 and .kind == "workflow_stage_timing" and
+    .producer == "benchkit" and .exp == "CASE0" and
+    .elapsed_clock == "monotonic" and
+    (.session_id | type == "string" and length > 0) and
+    (.stages[-1] |
+      .stage == "benchmark" and .tool == "none" and .status == "running" and
+      (.id | type == "string" and length > 0) and
+      (.started_at | type == "string" and length > 0) and
+      (.started_monotonic_ns | type == "number" and . > 0) and
+      (has("finished_at") | not) and (has("elapsed_seconds") | not) and
+      (has("exit_code") | not))
+  ' "$file" >/dev/null || return 99
+  cp "$file" "${TMP_DIR}/running.json" || return 99
+  printf 'command stdout\n'
+  printf 'command stderr\n' >&2
+  return "$1"
+}
+
+for route in direct file; do
+  for expected_status in 0 7 124; do
+    args=()
+    if [ "$route" = file ]; then
+      args=(--log "${TMP_DIR}/command.log")
+    fi
+    actual_status=0
+    output=$(bk_run "${args[@]}" -- probe_command "$expected_status" \
+      2>"${TMP_DIR}/stderr.log") || actual_status=$?
+    test "$actual_status" -eq "$expected_status"
+    if [ "$route" = direct ]; then
+      test "$output" = 'command stdout'
+      grep -Fxq 'command stderr' "${TMP_DIR}/stderr.log"
+    else
+      test -z "$output"
+      test "$(cat "${TMP_DIR}/command.log")" = $'command stdout\ncommand stderr'
+    fi
+    TIMING_FILE=$(timing_file CASE0)
+    jq -e --argjson status "$expected_status" --slurpfile running "${TMP_DIR}/running.json" '
+      .stages[-1] as $finished | $running[0].stages[-1] as $started |
+      .stages[0:-1] == $running[0].stages[0:-1] and
+      $finished.id == $started.id and $finished.started_at == $started.started_at and
+      $finished.stage == $started.stage and $finished.tool == $started.tool and
+      $finished.exit_code == $status and
+      $finished.status == (if $status == 0 then "completed" else "failed" end) and
+      ($finished.finished_at | type == "string" and length > 0) and
+      ($finished.elapsed_seconds | type == "number" and . >= 0)
+    ' "$TIMING_FILE" >/dev/null
+  done
+done
+cp "$TIMING_FILE" "${TMP_DIR}/known.json"
+
+# Absolute context survives cwd changes, substitutions, child shells and pipes.
+(
+  cd "${TMP_DIR}/nested/work"
+  bk_run -- printf 'pipeline output\n' | tee "${TMP_DIR}/pipeline.log" >/dev/null
+  test "$(bk_run -- pwd)" = "$PWD"
+  bash -c 'source "$1/scripts/bk_functions.sh"; bk_run -- true' bash "$REPO_DIR"
+  test ! -d results
+)
+test "$(cat "${TMP_DIR}/pipeline.log")" = 'pipeline output'
+jq -e '(.stages | length) == 9' "$TIMING_FILE" >/dev/null
+actual_status=0
+bk_run -- bash -c 'exit 7' | cat >/dev/null || actual_status=$?
+test "$actual_status" -eq 7
+test "$(bk_run -- printf 'two timed pipeline commands' | bk_run -- cat)" = 'two timed pipeline commands'
+(
+  set +e
+  bk_run -- false
+  test "$?" -eq 1
+  case "$-" in *e*) exit 1 ;; esac
+)
+
+for tool in nsys ncu fapp; do
+  for phase in collect export plan; do
+    bk_profile_execute --tool "$tool" --phase "$phase" --profile 'sample profile' \
+      --log "${TMP_DIR}/profile.log" -- printf '%s\n' "$tool:$phase"
+    test "$(cat "${TMP_DIR}/profile.log")" = "$tool:$phase"
+    jq -e --arg tool "$tool" --arg phase "$phase" '
+      .stages[-1] | .tool == $tool and .stage == $phase and
+      .profile == "sample profile" and .status == "completed" and .exit_code == 0
+    ' "$TIMING_FILE" >/dev/null
+  done
+done
+actual_status=0
+bk_profile_execute --tool ncu --phase collect -- bash -c 'exit 23' || actual_status=$?
+test "$actual_status" -eq 23
+jq -e '.stages[-1] | .status == "failed" and .exit_code == 23' "$TIMING_FILE" >/dev/null
+
+bk_generate_ncu_plan \
+  --nsys-csv "${REPO_DIR}/scripts/tests/fixtures/nsys_cuda_gpu_kern_sum.csv" \
+  --out-discovery "${RESULTS_DIR}/kernel_discovery.json" \
+  --out-plan "${RESULTS_DIR}/ncu_plan.json" --top-k 1 >/dev/null
+jq -e '.profiles | length == 1' "${RESULTS_DIR}/ncu_plan.json" >/dev/null
+jq -e '.stages[-1] | .stage == "plan" and .tool == "ncu" and .exit_code == 0' \
+  "$TIMING_FILE" >/dev/null
+actual_status=0
+bk_generate_ncu_plan --nsys-csv "${TMP_DIR}/missing.csv" \
+  --out-plan "${TMP_DIR}/missing-plan.json" >"${TMP_DIR}/bad-plan.log" 2>&1 || actual_status=$?
+test "$actual_status" -ne 0
+jq -e --argjson status "$actual_status" '.stages[-1] |
+  .stage == "plan" and .tool == "ncu" and .status == "failed" and .exit_code == $status' \
+  "$TIMING_FILE" >/dev/null
+
+# Interrupt the shell, not just its child: only the persisted start may remain.
+actual_status=0
+bash -c '
+  source "$1/scripts/bk_functions.sh"
+  trap "exit 143" TERM
+  bk_run -- bash -c '\''kill -TERM "$PPID"'\''
+' bash "$REPO_DIR" >"${TMP_DIR}/interrupted.log" 2>&1 || actual_status=$?
+test "$actual_status" -eq 143
+jq -e '.stages[-1] | .status == "running" and has("started_monotonic_ns") and
+  (has("finished_at") | not) and (has("elapsed_seconds") | not) and
+  (has("exit_code") | not)' "$TIMING_FILE" >/dev/null
+
+# Recorder unavailability must neither clobber known records nor mask status.
+cp "$TIMING_FILE" "${TMP_DIR}/before-unavailable.json"
+for expected_status in 0 7 124; do
+  actual_status=0
+  output=$(PYTHON_BIN="${TMP_DIR}/missing-python" bk_run -- \
+    bash -c 'printf "plan.json\n"; exit "$1"' bash "$expected_status" \
+    2>"${TMP_DIR}/unavailable.log") || actual_status=$?
+  test "$actual_status" -eq "$expected_status"
+  test "$output" = plan.json
+done
+cmp "$TIMING_FILE" "${TMP_DIR}/before-unavailable.json"
+
+timing_python_without_finish() {
+  local arg
+  for arg in "$@"; do
+    if [ "$arg" = finish ]; then
+      return 9
+    fi
+  done
+  "$PYTHON_FOR_TEST" "$@"
+}
+PYTHON_FOR_TEST="$PYTHON_BIN"
+for expected_status in 0 7 124; do
+  actual_status=0
+  PYTHON_BIN=timing_python_without_finish bk_run -- bash -c 'exit "$1"' bash "$expected_status" \
+    2>"${TMP_DIR}/finish-unavailable.log" || actual_status=$?
+  test "$actual_status" -eq "$expected_status"
+  jq -e '.stages[-1] | .status == "running" and
+    (has("exit_code") | not) and (has("elapsed_seconds") | not)' "$TIMING_FILE" >/dev/null
+done
+cp "$TIMING_FILE" "${TMP_DIR}/before-unavailable.json"
+jq -e --slurpfile known "${TMP_DIR}/known.json" '
+  .session_id == $known[0].session_id and
+  .stages[0:($known[0].stages | length)] == $known[0].stages and
+  ([.stages[].id] | unique | length) == (.stages | length)
+' "$TIMING_FILE" >/dev/null
+
+# Different experiments in the same results directory must not replace CASE0.
+bk_run_context --results-dir "$RESULTS_DIR" --exp CASE1
+bk_run -- true
+CASE1_FILE=$(timing_file CASE1)
+test "$CASE1_FILE" != "$TIMING_FILE"
+cmp "$TIMING_FILE" "${TMP_DIR}/before-unavailable.json"
+bk_run_context --results-dir "$RESULTS_DIR" --exp ''
+bk_run -- true
+manifest > "${TMP_DIR}/manifest.json"
+jq -e '
+  .schema_version == 1 and (.observations | length) == 3 and
+  ([.observations[].id] | unique | length) == 3 and
+  all(.observations[];
+    .kind == "workflow-stage-timing" and .producer == "benchkit" and
+    .format == "workflow_stage_timing/v1" and
+    .artifact.type == "file_reference" and
+    (.artifact.path | test("^results/workflow_timing_[a-zA-Z0-9]+\\.json$"))) and
+  any(.observations[]; (has("result_exp") | not))
+' "${TMP_DIR}/manifest.json" >/dev/null
+"$PYTHON_BIN" - "$RESULTS_DIR" "${TMP_DIR}/manifest.json" <<'PY'
+import json
+import pathlib
+import sys
+
+root = pathlib.Path(sys.argv[1])
+for observation in json.loads(pathlib.Path(sys.argv[2]).read_text())["observations"]:
+    stages = json.loads((root / pathlib.Path(observation["artifact"]["path"]).name).read_text())["stages"]
+    assert observation["summary"] == {
+        "stage_count": len(stages),
+        "completed_count": sum(stage["status"] == "completed" for stage in stages),
+        "failed_count": sum(stage["status"] == "failed" for stage in stages),
+        "unfinished_count": sum(stage["status"] not in {"completed", "failed"} for stage in stages),
+    }
+PY
+test ! -e "${RESULTS_DIR}/timing_observations.json"
+test ! -e "${RESULTS_DIR}/.timing_observation_items.jsonl"
+
+# Result emission must discover common artifacts without app registration.
+cat > "${RESULTS_DIR}/result" <<'EOF'
+FOM:1.25 FOM_unit:s Exp:CASE0 node_count:1 numproc_node:1 nthreads:1
+FOM:2.50 FOM_unit:s Exp:CASE1 node_count:1 numproc_node:1 nthreads:1
+EOF
+cat > "${RESULTS_DIR}/pipeline_timing.json" <<'EOF'
+{"build_time":12,"queue_time":0,"run_time":34}
+EOF
+(
+  cd "$TMP_DIR"
+  bash "${REPO_DIR}/scripts/result.sh" demoapp DemoSystem cross build_job run_job 42 >/dev/null
+)
+jq -e '(.timing_observations.observations | length) == 2' \
+  "${RESULTS_DIR}/result0.json" >/dev/null
+
+# Existing QWS detailed timing stays alongside the common workflow records.
+cat > "${RESULTS_DIR}/qws_timing_CASE0.json" <<'EOF'
+{"schema_version":1,"kind":"qws_timing_observation","producer":"qws","exp":"CASE0","timers":[]}
+EOF
+cat > "${RESULTS_DIR}/timing_observations.json" <<'EOF'
+{"schema_version":1,"observations":[{"id":"qws-case0","kind":"detailed-timing","producer":"qws","format":"qws_timing_observation/v1","result_exp":"CASE0","artifact":{"type":"file_reference","path":"results/qws_timing_CASE0.json"},"summary":{"timer_count":0}}]}
+EOF
+cp "${RESULTS_DIR}/timing_observations.json" "${TMP_DIR}/qws-manifest.json"
+(
+  cd "$TMP_DIR"
+  bash "${REPO_DIR}/scripts/result.sh" qws DemoSystem cross build_job run_job 42 >/dev/null
+)
+cmp "${RESULTS_DIR}/timing_observations.json" "${TMP_DIR}/qws-manifest.json"
+for index in 0 1; do
+  jq -e --arg exp "CASE$index" --argjson count "$((3 - index))" '
+    .Exp == $exp and
+    (.timing_observations.observations | length) == $count and
+    all(.timing_observations.observations[]; (.result_exp // $exp) == $exp) and
+    ([.timing_observations.observations[] | select(.producer == "benchkit")] | length) == 2 and
+    .pipeline_timing.build_time == 12 and .pipeline_timing.queue_time == 0 and
+    .pipeline_timing.run_time == 34 and (has("fom_breakdown") | not)
+  ' "${RESULTS_DIR}/result${index}.json" >/dev/null
+done
+jq -e '.FOM == "1.25" and any(.timing_observations.observations[]; .id == "qws-case0")' \
+  "${RESULTS_DIR}/result0.json" >/dev/null
+jq -e '.FOM == "2.50"' "${RESULTS_DIR}/result1.json" >/dev/null
+
+# Exercise the real sender; curl alone is mocked, so no network is possible.
+cat > "${TMP_DIR}/bin/curl" <<'EOF'
+#!/bin/bash
+set -euo pipefail
+printf '%s\n' "$*" >> "$CURL_LOG"
+case "$*" in
+  */api/ingest/result*) printf '%s\n' '{"id":"11111111-2222-3333-4444-555555555555","timestamp":"20260101_000000"}' ;;
+  *) printf '%s\n' '{"status":"uploaded"}' ;;
+esac
+EOF
+chmod +x "${TMP_DIR}/bin/curl"
+touch "${TMP_DIR}/client.crt" "${TMP_DIR}/client.key"
+(
+  cd "$TMP_DIR"
+  export PATH="${TMP_DIR}/bin:$PATH" CURL_LOG="${TMP_DIR}/curl.log"
+  export RESULT_SERVER=https://example.invalid
+  export RESULT_SERVER_CLIENT_CERT="${TMP_DIR}/client.crt"
+  export RESULT_SERVER_CLIENT_KEY="${TMP_DIR}/client.key"
+  bash "${REPO_DIR}/scripts/result_server/send_results.sh" >"${TMP_DIR}/sender.log" 2>&1
+)
+while IFS= read -r artifact; do
+  grep -F 'measurement-artifact' "${TMP_DIR}/curl.log" | grep -Fq "artifact_path=$artifact"
+done < <(jq -r '.observations[].artifact.path' "${TMP_DIR}/manifest.json")
+grep -F 'measurement-artifact' "${TMP_DIR}/curl.log" | grep -Fq 'artifact_path=results/qws_timing_CASE0.json'
+
+# GENESIS uses its actual launch builders with fake host/container profilers.
+cat > "${TMP_DIR}/bin/ncu" <<'EOF'
+#!/bin/bash
+set -euo pipefail
+if [ "${1:-}" = --import ]; then
+  printf 'metric,value\nfake,1\n'
+  exit 0
+fi
+output=''
+while [ "$#" -gt 0 ]; do
+  case "$1" in
+    -o) output="$2"; shift 2 ;;
+    --target-processes|--set|--launch-count|--kernel-name-base|--kernel-name|--launch-skip) shift 2 ;;
+    --nvtx) shift ;;
+    *) break ;;
+  esac
+done
+test -n "$output"
+mkdir -p "$(dirname "$output")"
+printf 'fake report\n' > "${output}.ncu-rep"
+"$@"
+EOF
+cat > "${TMP_DIR}/bin/nsys" <<'EOF'
+#!/bin/bash
+set -euo pipefail
+mode="$1"
+shift
+output=''
+while [ "$#" -gt 0 ]; do
+  case "$1" in
+    -o|--output) output="$2"; shift 2 ;;
+    --report|--format) shift 2 ;;
+    --*) shift ;;
+    *) break ;;
+  esac
+done
+test -n "$output"
+if [ "$mode" = profile ]; then
+  "$@"
+  printf 'fake report\n' > "${output}.nsys-rep"
+else
+  cp "$GENESIS_TEST_CSV" "$output"
+fi
+EOF
+cat > "${TMP_DIR}/bin/apptainer" <<'EOF'
+#!/bin/bash
+set -euo pipefail
+printf '%s\n' "$*" >> "$GENESIS_CONTAINER_LOG"
+test "$1" = exec
+shift
+while [ "$#" -gt 0 ]; do
+  case "$1" in
+    --nv) shift ;;
+    --bind|--pwd) shift 2 ;;
+    *) shift; break ;;
+  esac
+done
+# Preserve the fake tool PATH through GENESIS's container login-shell payload.
+if [ "${1:-}" = bash ] && [ "${2:-}" = -lc ]; then
+  shift 2
+  exec bash -c "$@"
+fi
+exec "$@"
+EOF
+chmod +x "${TMP_DIR}/bin/ncu" "${TMP_DIR}/bin/nsys" "${TMP_DIR}/bin/apptainer"
+for mode in host container; do
+  (
+    source "${REPO_DIR}/programs/genesis/profile.sh"
+    export PATH="${TMP_DIR}/bin:$PATH"
+    export GENESIS_TEST_CSV="${REPO_DIR}/scripts/tests/fixtures/nsys_cuda_gpu_kern_sum.csv"
+    export GENESIS_CONTAINER_LOG="${TMP_DIR}/container.log"
+    export BK_GENESIS_NCU_NSTEPS=off BK_GENESIS_NCU_PLAN_TOP_K=1
+    export GENESIS_BENCHKIT_ROOT="$REPO_DIR"
+    SCRIPT_DIR="$REPO_DIR"
+    resultsdir="${TMP_DIR}/genesis-${mode}/results"
+    RESULTS_DIR="$resultsdir"
+    header=sample
+    mkdir -p "$resultsdir"
+    cd "${TMP_DIR}/genesis-${mode}"
+    bk_run_context --results-dir "$resultsdir" --exp "GENESIS-${mode}"
+    app=(bash -c 'printf "dynamics = 1.25\n"' bash sample.inp)
+    if [ "$mode" = container ]; then
+      app=(apptainer exec --nv --bind "$PWD:$PWD" --pwd "$PWD" sample.sif "${app[@]}")
+    fi
+    bk_run -- "${app[@]}" >/dev/null
+    plan=$(genesis_generate_ncu_plan single "${app[@]}" 2>discovery.log)
+    test -s "$plan"
+    genesis_run_ncu_profile sample sample 'regex:.*sample.*' 0 1 single '' '{}' \
+      "${app[@]}" >profile.log 2>&1
+    jq -e '
+      .producer == "benchkit" and
+      any(.stages[]; .stage == "benchmark" and .tool == "none") and
+      any(.stages[]; .stage == "collect" and .tool == "nsys") and
+      any(.stages[]; .stage == "export" and .tool == "nsys") and
+      any(.stages[]; .stage == "plan" and .tool == "ncu") and
+      ([.stages[] | select(.stage == "collect" and .tool == "ncu")] | length) == 1 and
+      ([.stages[] | select(.stage == "export" and .tool == "ncu")] | length) == 2 and
+      all(.stages[]; .status == "completed" and .exit_code == 0)
+    ' "$resultsdir"/workflow_timing_*.json >/dev/null
+    test -f "$resultsdir/padata_sample.tgz"
+    test ! -e "$resultsdir/timing_observations.json"
+  )
+done
+grep -Fq 'nsys profile' "${TMP_DIR}/container.log"
+grep -Fq 'nsys stats' "${TMP_DIR}/container.log"
+grep -Fq 'ncu --import' "${TMP_DIR}/container.log"
+
+# A new session excludes previous-run files without deleting their evidence.
+cp "$TIMING_FILE" "${TMP_DIR}/previous-session.json"
+(
+  unset _BK_WORKFLOW_SESSION_ID
+  source "${REPO_DIR}/scripts/bk_functions.sh"
+  bk_run_context --results-dir "$RESULTS_DIR" --exp NEXT
+  manifest | jq -e '.observations == []' >/dev/null
+  bk_run -- true
+  manifest | jq -e '(.observations | length) == 1 and
+    .observations[0].result_exp == "NEXT"' >/dev/null
+)
+cmp "$TIMING_FILE" "${TMP_DIR}/previous-session.json"
+
+# Failed context publication cannot leave an earlier session active for emission.
+(
+  unset _BK_WORKFLOW_SESSION_ID
+  source "${REPO_DIR}/scripts/bk_functions.sh"
+  PYTHON_BIN="${TMP_DIR}/missing-python" \
+    bk_run_context --results-dir "$RESULTS_DIR" --exp UNAVAILABLE \
+    2>"${TMP_DIR}/context-unavailable.log"
+  manifest | jq -e '.schema_version == 1 and .observations == []' >/dev/null
+  cd "$TMP_DIR"
+  bash "${REPO_DIR}/scripts/result.sh" qws DemoSystem cross build_job run_job 42 >/dev/null
+  jq -e '[.timing_observations.observations[] | .id] == ["qws-case0"]' \
+    results/result0.json >/dev/null
+  jq -e '(.timing_observations.observations // []) == []' results/result1.json >/dev/null
+)
+cmp "$TIMING_FILE" "${TMP_DIR}/previous-session.json"
+
+echo "workflow timing tests passed"