diff --git a/CHANGELOG.md b/CHANGELOG.md
index eb12db95..7ef89fa0 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -17,7 +17,10 @@ Changes are grouped as follows
 ### Added
 * In the `unstable` package: New action status `completed_with_errors` in `ActionStatus`, reported when an action finishes but non-fatal (`error`-level) errors were reported during the run.
 
+* In the `unstable` package: New `Extractor.get_log_upload_data_set_id()` hook that tells the built-in `fetch_logs` action which data set to upload log files into. Resolution order: integration "data domains" field (not available yet, marked TODO), then the extractor's own configured data set, then no data set. The default uses `application_config.data_set` when the config derives from `WithDataSetId` and otherwise returns `None`; extractors that keep their data set elsewhere (for example the File and DB extractors) should override it in their own repositories.
+
 ### Changed
+* In the `unstable` package: `fetch_logs` now uploads a new copy of the log files on every invocation instead of overwriting: the file external ID is now `extractor-logs-<integration>-<date>-<action external id>` (the `extractor-logs-<integration>-<date>` prefix is unchanged). The per-file status `skipped` for a date without a log file is now called `missing`, matching the existing `missing_files` counter (all counter keys are unchanged). When no file was uploaded at all the action now ends as `completed_with_errors` with the message "No log files were uploaded ..." instead of `succeeded`. Custom actions can end as `completed_with_errors` with `ActionContext.set_result(..., completed_with_errors=True)`.
 * In the `unstable` package: `Stop <task>` actions now stay `running` until the task has actually ended and only then report `succeeded`. If the task has not ended within 300 seconds (`STOP_TASK_ACTION_TIMEOUT_SECONDS`) the action is reported as `failed` ("Task '<name>' did not stop within 300 seconds"); a waiting Stop is reported as `canceled` on extractor shutdown.
 
 ### Fixed
diff --git a/cognite/extractorutils/unstable/core/_log_upload_action.py b/cognite/extractorutils/unstable/core/_log_upload_action.py
index 9868f3d2..06dd889d 100644
--- a/cognite/extractorutils/unstable/core/_log_upload_action.py
+++ b/cognite/extractorutils/unstable/core/_log_upload_action.py
@@ -1,5 +1,15 @@
-"""Built-in ``fetch_logs`` action: streams rotated log files to CDF Files."""
+"""
+Built-in ``fetch_logs`` action: streams rotated log files to CDF Files.
 
+Every invocation uploads a new copy of each file (nothing is overwritten): the CDF external ID is
+``extractor-logs-<integration>-<log date>-<action external id>``. The files go into the data set returned by
+``Extractor.get_log_upload_data_set_id`` (none by default).
+
+Final status: ``failed`` for invalid parameters, ``completed_with_errors`` when no file at all was uploaded
+(everything missing, too large or failed), ``succeeded`` otherwise, including when only some days had a file.
+"""
+
+import hashlib
 import json
 import logging
 from collections import Counter
@@ -7,7 +17,7 @@
 from datetime import date, timedelta, timezone
 from datetime import datetime as dt
 from pathlib import Path
-from typing import BinaryIO, Literal
+from typing import Any, BinaryIO, Literal
 
 from cognite.client import CogniteClient
 
@@ -81,23 +91,23 @@ def _build_candidate_files(
     today: date,
 ) -> tuple[list[LogFileCandidate], list[date]]:
     """
-    Enumerate log files for [start_date, end_date] and partition into candidates vs skipped.
+    Enumerate log files for [start_date, end_date] and partition into candidates vs missing.
 
     Rotated files follow the naming convention ``<base_path>.YYYY-MM-DD``.
     The live file (``base_path``) is used for ``today``; all other dates use the rotated name.
 
-    A file is skipped when it does not exist, is empty, or is inaccessible (any ``OSError``).
+    A file is missing when it does not exist, is empty, or is inaccessible (any ``OSError``).
     This covers the brief rotation race window: if the action is dispatched right after midnight
     before ``TimedRotatingFileHandler`` has renamed ``file.log`` to ``file.log.YYYY-MM-DD``,
-    the rotated file will not be found and that date will appear in ``skipped``. Retrying after
+    the rotated file will not be found and that date will be reported as ``missing``. Retrying after
     rotation completes (within seconds) will pick it up.
 
     Returns:
-        (candidates, skipped): candidates are files that exist and have content;
-        skipped contains dates for which no usable file was found.
+        (candidates, missing): candidates are files that exist and have content;
+        missing contains dates for which no usable file was found.
     """
     candidates: list[LogFileCandidate] = []
-    skipped: list[date] = []
+    missing: list[date] = []
 
     current = start_date
     while current <= end_date:
@@ -111,30 +121,55 @@ def _build_candidate_files(
         try:
             size = path.stat().st_size
         except OSError as e:
-            _logger.warning("fetch_logs: skipping %s for %s — %s", path, current, e)
-            skipped.append(current)
+            _logger.warning("fetch_logs: no usable log file %s for %s — %s", path, current, e)
+            missing.append(current)
         else:
             if size > 0:
                 candidates.append(LogFileCandidate(log_date=current, path=path, is_current=is_current))
             else:
-                skipped.append(current)
+                missing.append(current)
 
         current += timedelta(days=1)
 
-    return candidates, skipped
+    return candidates, missing
+
 
+MAX_EXTERNAL_ID_LENGTH = 255
+"""CDF's limit for file external IDs."""
+
+
+def _file_external_id(integration_external_id: str, log_date: date, run_id: str) -> str:
+    """
+    Build the CDF external ID of one uploaded log file.
 
-def _file_external_id(integration_external_id: str, log_date: date) -> str:
-    return f"extractor-logs-{integration_external_id}-{log_date.isoformat()}"
+    ``run_id`` (the external ID of the action invocation) makes the ID unique per run so repeated runs never
+    overwrite each other, while the ``extractor-logs-<integration>-<date>`` prefix keeps the files findable. IDs
+    are deterministic for a given run. If the result would exceed CDF's limit, ``run_id`` is replaced by a hash of
+    itself and, as a last resort, the integration part is shortened.
+    """
+    suffix = f"{log_date.isoformat()}-{run_id}"
+    external_id = f"extractor-logs-{integration_external_id}-{suffix}"
+    if len(external_id) <= MAX_EXTERNAL_ID_LENGTH:
+        return external_id
+    suffix = f"{log_date.isoformat()}-{hashlib.sha256(run_id.encode('utf-8')).hexdigest()[:16]}"
+    room = MAX_EXTERNAL_ID_LENGTH - len("extractor-logs--") - len(suffix)
+    return f"extractor-logs-{integration_external_id[:room]}-{suffix}"
 
 
 def _upload_candidate(
     candidate: LogFileCandidate,
     integration_external_id: str,
     cdf_client: CogniteClient,
+    *,
+    run_id: str,
+    data_set_id: int | None = None,
 ) -> _FileUploadResult:
-    """Upload one candidate log file to CDF Files. Returns a result regardless of success or failure."""
-    external_id = _file_external_id(integration_external_id, candidate.log_date)
+    """
+    Upload one candidate log file to CDF Files as a new file. Returns a result regardless of success or failure.
+
+    The file is never overwritten: ``run_id`` makes the external ID unique per ``fetch_logs`` invocation.
+    """
+    external_id = _file_external_id(integration_external_id, candidate.log_date, run_id)
 
     try:
         actual_size = candidate.path.stat().st_size
@@ -157,6 +192,8 @@ def _upload_candidate(
             size_bytes=actual_size,
         )
 
+    # Only send data_set_id when there is one, so extractors without a data set upload exactly as before.
+    extra_upload_args: dict[str, Any] = {} if data_set_id is None else {"data_set_id": data_set_id}
     try:
         with open(candidate.path, "rb") as f:
             reader: BinaryIO = BoundedReader(f, actual_size) if candidate.is_current else f  # type: ignore[assignment]
@@ -165,7 +202,7 @@ def _upload_candidate(
                 name=f"{external_id}.log",
                 external_id=external_id,
                 mime_type="text/plain",
-                overwrite=True,
+                **extra_upload_args,
             )
         _logger.info("fetch_logs: uploaded %s (%d bytes)", external_id, actual_size)
         return _FileUploadResult(
@@ -229,27 +266,32 @@ def fetch_logs_action(ctx: ActionContext) -> None:
             error_type="no_file_handler_configured",
         )
 
-    candidates, skipped_dates = _build_candidate_files(log_file_path, start_date, end_date, today)
+    candidates, missing_dates = _build_candidate_files(log_file_path, start_date, end_date, today)
     _logger.info(
-        "fetch_logs: %d candidate file(s) for %s to %s; %d date(s) skipped",
+        "fetch_logs: %d candidate file(s) for %s to %s; %d date(s) missing",
         len(candidates),
         start_date,
         end_date,
-        len(skipped_dates),
+        len(missing_dates),
     )
 
     integration_external_id = ctx.integration_external_id
     cdf_client = ctx.cdf_client
+    data_set_id = ctx.log_upload_data_set_id()
 
     total_candidates = len(candidates)
     upload_results: list[_FileUploadResult] = []
     for i, candidate in enumerate(candidates, 1):
-        upload_results.append(_upload_candidate(candidate, integration_external_id, cdf_client))
+        upload_results.append(
+            _upload_candidate(
+                candidate, integration_external_id, cdf_client, run_id=ctx.external_id, data_set_id=data_set_id
+            )
+        )
         ctx.report_progress(f"Uploading: {i}/{total_candidates} files complete")
 
     counts = Counter(r.status for r in upload_results)
 
-    # Per-file entries: upload results (sorted by date) + missing dates (skipped by candidate builder).
+    # Per-file entries: upload results (sorted by date) + missing dates (no usable file found by the candidate builder).
     files_list: list[dict[str, str]] = []
     for r in upload_results:
         entry: dict[str, str] = {"date": str(r.log_date), "status": r.status}
@@ -260,14 +302,17 @@ def fetch_logs_action(ctx: ActionContext) -> None:
         if r.error:
             entry["error"] = r.error
         files_list.append(entry)
-    files_list.extend({"date": str(skipped_date), "status": "skipped"} for skipped_date in skipped_dates)
+    files_list.extend({"date": str(missing_date), "status": "missing"} for missing_date in missing_dates)
     files_list.sort(key=lambda e: e["date"])
 
-    result_message = f"{counts['uploaded']} of {num_days} log files uploaded to CDF Files"
+    uploaded = counts["uploaded"]
+    result_message = f"{uploaded} of {num_days} log files uploaded to CDF Files"
+    if uploaded == 0:
+        result_message = f"No log files were uploaded ({result_message})"
     metadata: dict[str, str] = {
         "total_files": str(num_days),
         "uploaded_files": str(counts["uploaded"]),
-        "missing_files": str(len(skipped_dates)),
+        "missing_files": str(len(missing_dates)),
         "skipped_too_large_files": str(counts["skipped_too_large"]),
         "failed_files": str(counts["failed"]),
     }
@@ -290,4 +335,4 @@ def fetch_logs_action(ctx: ActionContext) -> None:
         )
         result_message += " (per-file detail omitted from result metadata — see extractor logs for full detail)"
 
-    ctx.set_result(result_message, metadata=metadata)
+    ctx.set_result(result_message, metadata=metadata, completed_with_errors=uploaded == 0)
diff --git a/cognite/extractorutils/unstable/core/actions.py b/cognite/extractorutils/unstable/core/actions.py
index f6cb50a6..07ca011a 100644
--- a/cognite/extractorutils/unstable/core/actions.py
+++ b/cognite/extractorutils/unstable/core/actions.py
@@ -50,6 +50,7 @@ def __init__(
         self.cancellation_token = cancellation_token
         self._result_message: str | None = None
         self._result_metadata: dict[str, str] | None = None
+        self._completed_with_errors = False
 
         self._logger = logging.getLogger(f"{self._extractor.EXTERNAL_ID}.action.{self._action.name.replace(' ', '')}")
 
@@ -83,8 +84,22 @@ def _new_error(
             task_name=task_name if task_name is not None else self._action.name,
         )
 
-    def set_result(self, message: str, *, metadata: dict[str, str] | None = None) -> None:
-        """Record the result for a successful action completion."""
+    def log_upload_data_set_id(self) -> int | None:
+        """The data set the ``fetch_logs`` action uploads into, see ``Extractor.get_log_upload_data_set_id``."""
+        return self._extractor.get_log_upload_data_set_id()
+
+    def set_result(
+        self, message: str, *, metadata: dict[str, str] | None = None, completed_with_errors: bool = False
+    ) -> None:
+        """
+        Record the result for an action completion.
+
+        Args:
+            message: The result message.
+            metadata: Optional result metadata.
+            completed_with_errors: Finish the action as ``completed_with_errors`` instead of ``succeeded``, for an
+                action that ran to the end but did not achieve its goal. Ignored if the action was cancelled.
+        """
         if self._result_message is not None:
             raise RuntimeError(
                 f"set_result() has already been called for this action invocation; "
@@ -92,6 +107,7 @@ def set_result(self, message: str, *, metadata: dict[str, str] | None = None) ->
             )
         self._result_message = message
         self._result_metadata = metadata
+        self._completed_with_errors = completed_with_errors
 
     def report_progress(self, message: str) -> None:
         """Queue an intermediate progress update while the action is still running."""
diff --git a/cognite/extractorutils/unstable/core/base.py b/cognite/extractorutils/unstable/core/base.py
index 09dddf85..cfdbf425 100644
--- a/cognite/extractorutils/unstable/core/base.py
+++ b/cognite/extractorutils/unstable/core/base.py
@@ -110,6 +110,7 @@ def my_task_function(self, task_context: TaskContext) -> None:
     ExtractorConfig,
     LogConsoleHandlerConfig,
     LogFileHandlerConfig,
+    WithDataSetId,
     create_state_store,
 )
 from cognite.extractorutils.unstable.core._dto import (
@@ -442,6 +443,28 @@ def __init_actions__(self) -> None:
         """
         pass
 
+    def get_log_upload_data_set_id(self) -> int | None:
+        """
+        Return the id of the CDF data set the built-in ``fetch_logs`` action uploads log files into, or ``None``.
+
+        Resolution order:
+
+        1. TODO(data-domains): the integration's "data domains" field. It does not exist in the Integrations API yet;
+           when it does, resolve it here and prefer it over everything below. Intentionally not implemented.
+        2. The extractor's own configured data set. By default this is ``application_config.data_set`` if the
+           application config derives from ``WithDataSetId``. Extractors whose config keeps the data set elsewhere
+           (for example per destination) should override this method.
+        3. ``None``: the files are uploaded without a data set.
+
+        Called on every ``fetch_logs`` invocation. An exception raised here (for example the data set cannot be
+        found) fails the action.
+        """
+        config = self.application_config
+        if isinstance(config, WithDataSetId):
+            data_set = config.get_data_set(self.cognite_client)
+            return data_set.id if data_set is not None else None
+        return None
+
     def add_action(self, action: CustomAction) -> None:
         """
         Register a custom action that Odin can invoke on this extractor.
@@ -1005,7 +1028,12 @@ def _handle_custom_action(self, action: Action) -> None:
 
             custom.target(ctx)
             filtered_metadata, oversized_fields = drop_oversized_metadata_fields(ctx._result_metadata)
-            completed_status = ActionStatus.canceled if action_token.is_cancelled else ActionStatus.succeeded
+            if action_token.is_cancelled:
+                completed_status = ActionStatus.canceled
+            elif ctx._completed_with_errors:
+                completed_status = ActionStatus.completed_with_errors
+            else:
+                completed_status = ActionStatus.succeeded
             if oversized_fields:
                 # The action itself ran to completion (or was cancelled) — only reporting the full
                 # result back to Odin is affected, because a metadata value is too large to send.
@@ -1014,7 +1042,10 @@ def _handle_custom_action(self, action: Action) -> None:
                 # bundles all pending updates together and requeues the whole batch on any rejection).
                 # Non-oversized fields are still reported, and the action's real outcome (succeeded or
                 # canceled) is preserved rather than being overwritten by this reporting issue.
-                outcome = "was canceled" if completed_status == ActionStatus.canceled else "completed successfully"
+                outcome = {
+                    ActionStatus.canceled: "was canceled",
+                    ActionStatus.completed_with_errors: "completed with errors",
+                }.get(completed_status, "completed successfully")
                 self._checkin_worker.queue_action_update(
                     ActionUpdate(
                         external_id=action.external_id,
diff --git a/tests/test_unstable/test_log_upload_action.py b/tests/test_unstable/test_log_upload_action.py
index 3b798e38..8a7c8b7a 100644
--- a/tests/test_unstable/test_log_upload_action.py
+++ b/tests/test_unstable/test_log_upload_action.py
@@ -286,18 +286,21 @@ def test_no_file_handler_reports_no_file_handler_configured() -> None:
     [True, False],
     ids=["files_present", "all_files_missing"],
 )
-def test_fetch_logs_action_with_file_handler_succeeds(tmp_path: Path, create_log_file: bool) -> None:
+def test_fetch_logs_action_with_file_handler_does_not_fail(tmp_path: Path, create_log_file: bool) -> None:
     log_path = tmp_path / "extractor.log"
     if create_log_file:
         (tmp_path / "extractor.log.2026-06-10").write_bytes(b"log data")
     extractor = _make_extractor(log_path=log_path)
     updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
     statuses = [u.status for u in updates]
-    assert ActionStatus.succeeded in statuses
     assert ActionStatus.failed not in statuses
+    if create_log_file:
+        assert ActionStatus.succeeded in statuses
+    else:
+        assert ActionStatus.completed_with_errors in statuses
 
 
-def test_valid_exact_max_days_range_succeeds_at_dispatch(tmp_path: Path) -> None:
+def test_valid_exact_max_days_range_is_accepted_at_dispatch(tmp_path: Path) -> None:
     # Exactly MAX_DATE_RANGE_DAYS is the boundary — one more would be rejected.
     start = date(2026, 6, 1)
     end = start + timedelta(days=MAX_DATE_RANGE_DAYS - 1)
@@ -305,7 +308,7 @@ def test_valid_exact_max_days_range_succeeds_at_dispatch(tmp_path: Path) -> None
     extractor = _make_extractor(log_path=log_path)
     updates = _dispatch(extractor, {"start_date": str(start), "end_date": str(end)})
     statuses = [u.status for u in updates]
-    assert ActionStatus.succeeded in statuses
+    assert ActionStatus.completed_with_errors in statuses  # no files exist, nothing is uploaded
     assert ActionStatus.failed not in statuses
 
 
@@ -341,7 +344,7 @@ def test_max_range_with_uploads_keeps_files_metadata_within_budget(tmp_path: Pat
 
 def test_files_metadata_omitted_when_still_oversized(tmp_path: Path) -> None:
     # Defense in depth: even with the compact schema, a pathological case (e.g. long per-file error
-    # messages) could still overflow the budget. The action must still succeed on the aggregate
+    # messages) could still overflow the budget. The action must still be reported on the aggregate
     # counts, gracefully omitting 'files' rather than producing an oversized result that Odin would
     # reject and retry forever.
     start = date(2026, 6, 1)
@@ -352,7 +355,7 @@ def test_files_metadata_omitted_when_still_oversized(tmp_path: Path) -> None:
     extractor = _make_extractor(log_path=log_path)
     extractor.cognite_client.files.upload_bytes.side_effect = RuntimeError("x" * 150)
     updates = _dispatch(extractor, {"start_date": str(start), "end_date": str(end)})
-    succeeded = next(u for u in updates if u.status == ActionStatus.succeeded)
+    succeeded = next(u for u in updates if u.status == ActionStatus.completed_with_errors)
     assert succeeded.result_metadata is not None
     assert succeeded.result_metadata["failed_files"] == str(MAX_DATE_RANGE_DAYS)
     assert "files" not in succeeded.result_metadata
@@ -378,7 +381,9 @@ def test_range_spanning_rotated_and_live_file(tmp_path: Path) -> None:
 def test_file_external_id_naming_convention() -> None:
     from cognite.extractorutils.unstable.core._log_upload_action import _file_external_id
 
-    assert _file_external_id("my-extractor", date(2026, 6, 1)) == "extractor-logs-my-extractor-2026-06-01"
+    assert (
+        _file_external_id("my-extractor", date(2026, 6, 1), "act-1") == "extractor-logs-my-extractor-2026-06-01-act-1"
+    )
 
 
 def test_upload_candidate_calls_cdf_upload_for_rotated_file(tmp_path: Path) -> None:
@@ -389,15 +394,16 @@ def test_upload_candidate_calls_cdf_upload_for_rotated_file(tmp_path: Path) -> N
     candidate = LogFileCandidate(log_date=date(2026, 6, 1), path=path, is_current=False)
     mock_client = MagicMock()
 
-    result = _upload_candidate(candidate, "my-extractor", mock_client)
+    result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "uploaded"
     assert result.size_bytes == path.stat().st_size
-    assert result.file_external_id == "extractor-logs-my-extractor-2026-06-01"
+    assert result.file_external_id == "extractor-logs-my-extractor-2026-06-01-act-1"
     mock_client.files.upload_bytes.assert_called_once()
     _, kwargs = mock_client.files.upload_bytes.call_args
-    assert kwargs["external_id"] == "extractor-logs-my-extractor-2026-06-01"
-    assert kwargs["overwrite"] is True
+    assert kwargs["external_id"] == "extractor-logs-my-extractor-2026-06-01-act-1"
+    assert "overwrite" not in kwargs
+    assert "data_set_id" not in kwargs
 
 
 def test_upload_candidate_current_day_uses_bounded_reader(tmp_path: Path) -> None:
@@ -409,7 +415,7 @@ def test_upload_candidate_current_day_uses_bounded_reader(tmp_path: Path) -> Non
     candidate = LogFileCandidate(log_date=_PAST_TODAY, path=path, is_current=True)
     mock_client = MagicMock()
 
-    result = _upload_candidate(candidate, "my-extractor", mock_client)
+    result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "uploaded"
     assert result.size_bytes == 300
@@ -427,7 +433,7 @@ def test_upload_candidate_rotated_file_does_not_use_bounded_reader(tmp_path: Pat
     candidate = LogFileCandidate(log_date=date(2026, 6, 1), path=path, is_current=False)
     mock_client = MagicMock()
 
-    _upload_candidate(candidate, "my-extractor", mock_client)
+    _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     _, kwargs = mock_client.files.upload_bytes.call_args
     assert not isinstance(kwargs["content"], BoundedReader)
@@ -449,7 +455,7 @@ def test_upload_candidate_current_day_bounded_reader_caps_to_stat_size(tmp_path:
     mock_stat_result = MagicMock()
     mock_stat_result.st_size = 300
     with patch("pathlib.Path.stat", return_value=mock_stat_result):
-        result = _upload_candidate(candidate, "my-extractor", mock_client)
+        result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "uploaded"
     assert result.size_bytes == 300
@@ -473,7 +479,7 @@ def capture_upload(*, content: BoundedReader, **kwargs: object) -> MagicMock:
     mock_client = MagicMock()
     mock_client.files.upload_bytes.side_effect = capture_upload
 
-    result = _upload_candidate(candidate, "my-extractor", mock_client)
+    result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "uploaded"
     assert consumed["length"] == len(file_content)
@@ -498,7 +504,7 @@ def test_upload_candidate_exceeds_max_size_returns_skipped_too_large(tmp_path: P
     mock_stat_result = MagicMock()
     mock_stat_result.st_size = oversized
     with patch("pathlib.Path.stat", return_value=mock_stat_result):
-        result = _upload_candidate(candidate, "my-extractor", mock_client)
+        result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "skipped_too_large"
     assert result.size_bytes == oversized
@@ -514,7 +520,7 @@ def test_upload_candidate_cdf_error_returns_failed(tmp_path: Path) -> None:
     mock_client = MagicMock()
     mock_client.files.upload_bytes.side_effect = RuntimeError("CDF unavailable")
 
-    result = _upload_candidate(candidate, "my-extractor", mock_client)
+    result = _upload_candidate(candidate, "my-extractor", mock_client, run_id="act-1")
 
     assert result.status == "failed"
     assert "CDF unavailable" in (result.error or "")
@@ -552,15 +558,18 @@ def test_fetch_logs_action_missing_files_reported_in_metadata(tmp_path: Path) ->
     assert succeeded.result_metadata["total_files"] == "2"
     files = json.loads(succeeded.result_metadata["files"])
     by_date = {f["date"]: f for f in files}
-    assert by_date["2026-06-10"]["status"] == "skipped"
+    assert by_date["2026-06-10"]["status"] == "missing"
     assert by_date["2026-06-11"]["status"] == "uploaded"
 
 
-def test_fetch_logs_action_all_files_missing_still_succeeds(tmp_path: Path) -> None:
+def test_fetch_logs_action_all_files_missing_completes_with_errors(tmp_path: Path) -> None:
     log_path = tmp_path / "extractor.log"
     extractor = _make_extractor(log_path=log_path)
     updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
-    succeeded = next(u for u in updates if u.status == ActionStatus.succeeded)
+    assert ActionStatus.succeeded not in [u.status for u in updates]
+    succeeded = next(u for u in updates if u.status == ActionStatus.completed_with_errors)
+    assert succeeded.result_message is not None
+    assert succeeded.result_message.startswith("No log files were uploaded")
     assert succeeded.result_metadata is not None
     assert succeeded.result_metadata["uploaded_files"] == "0"
     assert succeeded.result_metadata["missing_files"] == "1"
@@ -597,7 +606,7 @@ def test_fetch_logs_action_stat_failure_for_current_file_reports_failed(tmp_path
     ):
         updates = _dispatch(extractor, {"start_date": str(_PAST_TODAY), "end_date": str(_PAST_TODAY)})
 
-    succeeded = next(u for u in updates if u.status == ActionStatus.succeeded)
+    succeeded = next(u for u in updates if u.status == ActionStatus.completed_with_errors)
     assert succeeded.result_metadata is not None
     assert succeeded.result_metadata["failed_files"] == "1"
     assert succeeded.result_metadata["uploaded_files"] == "0"
@@ -606,7 +615,7 @@ def test_fetch_logs_action_stat_failure_for_current_file_reports_failed(tmp_path
     assert "permission denied" in files[0]["error"]
 
 
-def test_fetch_logs_action_upload_failure_still_succeeds(tmp_path: Path) -> None:
+def test_fetch_logs_action_upload_failure_completes_with_errors(tmp_path: Path) -> None:
     from unittest.mock import patch
 
     log_path = tmp_path / "extractor.log"
@@ -622,7 +631,7 @@ def test_fetch_logs_action_upload_failure_still_succeeds(tmp_path: Path) -> None
         ),
     ):
         updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
-    succeeded = next(u for u in updates if u.status == ActionStatus.succeeded)
+    succeeded = next(u for u in updates if u.status == ActionStatus.completed_with_errors)
     assert succeeded.result_metadata is not None
     assert succeeded.result_metadata["failed_files"] == "1"
     assert succeeded.result_metadata["uploaded_files"] == "0"
@@ -675,4 +684,115 @@ def test_upload_action_no_progress_updates_when_all_files_missing(tmp_path: Path
     updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
     progress_updates = [u for u in updates if u.status == ActionStatus.running and u.result_message is not None]
     assert progress_updates == []
-    assert any(u.status == ActionStatus.succeeded for u in updates)
+    assert any(u.status == ActionStatus.completed_with_errors for u in updates)
+
+
+# --- data set, per-run external IDs and outcome status (EDG-1037) ---
+
+
+def _upload_kwargs(extractor: TestExtractor) -> list[dict]:
+    return [c.kwargs for c in extractor.cognite_client.files.upload_bytes.call_args_list]
+
+
+def test_fetch_logs_passes_data_set_id_from_extractor_hook(tmp_path: Path) -> None:
+    (tmp_path / "extractor.log.2026-06-10").write_bytes(b"data")
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    extractor.get_log_upload_data_set_id = lambda: 4242  # type: ignore[method-assign]
+    _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
+    assert [k["data_set_id"] for k in _upload_kwargs(extractor)] == [4242]
+
+
+def test_fetch_logs_omits_data_set_id_by_default(tmp_path: Path) -> None:
+    (tmp_path / "extractor.log.2026-06-10").write_bytes(b"data")
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-10"})
+    kwargs = _upload_kwargs(extractor)
+    assert len(kwargs) == 1
+    assert "data_set_id" not in kwargs[0]
+
+
+def test_default_data_set_hook_uses_with_data_set_id_config() -> None:
+    from cognite.extractorutils.unstable.configuration.models import WithDataSetId
+
+    extractor = _make_extractor()
+    assert extractor.get_log_upload_data_set_id() is None  # TestConfig has no data set
+
+    class _Cfg(WithDataSetId):
+        pass
+
+    cfg = _Cfg.model_validate({"data-set": {"external-id": "ds"}})
+    extractor.application_config = cfg  # type: ignore[assignment]
+    extractor.cognite_client.data_sets.retrieve.return_value = MagicMock(id=77)
+    assert extractor.get_log_upload_data_set_id() == 77
+
+
+def test_fetch_logs_twice_for_same_day_uploads_two_different_files_without_overwrite(tmp_path: Path) -> None:
+    (tmp_path / "extractor.log.2026-06-10").write_bytes(b"data")
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    for external_id in ("act-1", "act-2"):
+        extractor._dispatch_single_action(
+            Action(
+                external_id=external_id,
+                action_name="fetch_logs",
+                status=ActionStatus.pending,
+                call_metadata={"start_date": "2026-06-10", "end_date": "2026-06-10"},
+            )
+        )
+    kwargs = _upload_kwargs(extractor)
+    ids = [k["external_id"] for k in kwargs]
+    assert ids == [
+        "extractor-logs-test-integration-2026-06-10-act-1",
+        "extractor-logs-test-integration-2026-06-10-act-2",
+    ]
+    assert all("overwrite" not in k for k in kwargs)
+
+
+def test_file_external_id_is_deterministic_and_within_cdf_limit() -> None:
+    from cognite.extractorutils.unstable.core._log_upload_action import MAX_EXTERNAL_ID_LENGTH, _file_external_id
+
+    d = date(2026, 6, 1)
+    assert _file_external_id("integ", d, "run") == _file_external_id("integ", d, "run")
+    long_id = _file_external_id("i" * 255, d, "r" * 255)
+    assert len(long_id) <= MAX_EXTERNAL_ID_LENGTH
+    assert long_id.startswith("extractor-logs-iii")
+    assert _file_external_id("i" * 255, d, "r" * 255) == long_id
+    assert _file_external_id("integ", d, "r" * 255) != _file_external_id("integ", d, "s" * 255)
+    assert len(_file_external_id("integ", d, "r" * 255)) <= MAX_EXTERNAL_ID_LENGTH
+
+
+def test_fetch_logs_partial_upload_is_succeeded(tmp_path: Path) -> None:
+    (tmp_path / "extractor.log.2026-06-11").write_bytes(b"data")  # 2026-06-10 is missing
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-11"})
+    final = updates[-1]
+    assert final.status == ActionStatus.succeeded
+    assert "No log files were uploaded" not in (final.result_message or "")
+
+
+def test_fetch_logs_nothing_uploaded_completes_with_errors_and_message(tmp_path: Path) -> None:
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    updates = _dispatch(extractor, {"start_date": "2026-06-10", "end_date": "2026-06-11"})
+    final = updates[-1]
+    assert final.status == ActionStatus.completed_with_errors
+    assert final.result_message == "No log files were uploaded (0 of 2 log files uploaded to CDF Files)"
+    assert final.result_metadata is not None
+    assert final.result_metadata["missing_files"] == "2"
+
+
+def test_fetch_logs_invalid_range_still_failed(tmp_path: Path) -> None:
+    extractor = _make_extractor(log_path=tmp_path / "extractor.log")
+    updates = _dispatch(extractor, {"start_date": "2026-06-11", "end_date": "2026-06-10"})
+    assert updates[-1].status == ActionStatus.failed
+    assert updates[-1].result_metadata == {"error_type": "invalid_date_range"}
+
+
+def test_set_result_completed_with_errors_is_ignored_when_cancelled() -> None:
+    extractor = _make_extractor()
+
+    def target(ctx: ActionContext) -> None:
+        ctx.set_result("nothing done", completed_with_errors=True)
+        ctx.cancellation_token.cancel()
+
+    extractor.add_action(CustomAction(name="custom", target=target))
+    extractor._dispatch_single_action(Action(external_id="a", action_name="custom", status=ActionStatus.pending))
+    assert _queued_updates(extractor)[-1].status == ActionStatus.canceled