From 419546aef583ab7e3bc942d62755e1d8161d28ff Mon Sep 17 00:00:00 2001
From: Karl Weinmeister <kweinmeister@google.com>
Date: Sun, 13 Sep 2026 11:16:43 -0500
Subject: [PATCH 1/4] fix(harbor): align dual-arm task naming and suffix
 stripping with skill standards

Signed-off-by: Karl Weinmeister <kweinmeister@google.com>
---
 src/skillevaluator/tier3/harbor/adapter.py   | 14 ++++-
 src/skillevaluator/tier3/harbor/collector.py | 14 ++++-
 src/skillevaluator/tier3/harbor/runner.py    |  5 ++
 tests/test_tier3_public_runtime.py           | 60 ++++++++++++++++++++
 4 files changed, 89 insertions(+), 4 deletions(-)

diff --git a/src/skillevaluator/tier3/harbor/adapter.py b/src/skillevaluator/tier3/harbor/adapter.py
index cbd0518d..d7265625 100644
--- a/src/skillevaluator/tier3/harbor/adapter.py
+++ b/src/skillevaluator/tier3/harbor/adapter.py
@@ -1794,6 +1794,7 @@ def _write_task_toml(
     pre_agent_setup: list[str] | None = None,
     task_resources: dict[str, int] | None = None,
     agent_workdir: str | None = None,
+    arm_suffix: str = "",
 ) -> None:
     entry_id = entry.get("id", "unknown")
     expected_skill = entry.get("expected_skill") or "none"
@@ -1809,10 +1810,11 @@ def _write_task_toml(
     storage_mb = _task_resource_value(task_resources, "storage_mb", 2048)
     workdir_line = f"workdir = {_toml_quote(agent_workdir)}\n" if agent_workdir else ""
 
+    task_name = f"nvidia/skillevaluator-{entry_id}{arm_suffix}"
     content = f"""schema_version = "1.3"
 
 [task]
-name = {_toml_quote(f"nvidia/skillevaluator-{entry_id}")}
+name = {_toml_quote(task_name)}
 description = {_toml_quote(f"Skill evaluation task for {expected_skill}")}
 
 [metadata]
@@ -4579,6 +4581,7 @@ def _stage_native_harbor_tasks_into(
     task_resources: dict[str, int] | None = None,
     agent_workdir: str | None = None,
     baseline_aliases_prevalidated: bool = False,
+    arm_suffix: str = "",
 ) -> list[Path]:
     """Build native Harbor tasks inside a private, caller-owned directory.
 
@@ -4608,6 +4611,7 @@ def _stage_native_harbor_tasks_into(
         _native_task_workdir(source_task_dir)
     _ = task_resources
     _ = agent_workdir
+    _ = arm_suffix
 
     if output_dir.exists():
         shutil.rmtree(output_dir)
@@ -4767,6 +4771,7 @@ def stage_native_harbor_tasks(
     agent_workdir: str | None = None,
     evaluator_skill_path: Path | None = None,
     _baseline_alias_validation: _BaselineAliasValidation | None = None,
+    arm_suffix: str = "",
 ) -> list[Path]:
     """Stage native tasks privately, then publish one exact output snapshot."""
 
@@ -4791,6 +4796,7 @@ def stage_native_harbor_tasks(
                 agent_workdir=agent_workdir,
                 evaluator_skill_path=private_skill_path,
                 _baseline_alias_validation=_baseline_alias_validation,
+                arm_suffix=arm_suffix,
             )
 
     baseline_aliases_prevalidated = False
@@ -4857,6 +4863,7 @@ def stage_native_harbor_tasks(
             task_resources=task_resources,
             agent_workdir=agent_workdir,
             baseline_aliases_prevalidated=baseline_aliases_prevalidated,
+            arm_suffix=arm_suffix,
         )
         relative_tasks = [task.relative_to(private_output) for task in private_tasks]
         if output_requires_provenance:
@@ -4915,6 +4922,7 @@ def _generate_harbor_tasks_into(
     task_resources: dict[str, int] | None = None,
     agent_workdir: str | None = None,
     baseline_aliases_prevalidated: bool = False,
+    arm_suffix: str = "",
 ) -> list[Path]:
     """Generate Harbor task directories inside a private output directory.
 
@@ -5011,6 +5019,7 @@ def _generate_harbor_tasks_into(
             pre_agent_setup=pre_agent_setup,
             task_resources=task_resources,
             agent_workdir=agent_workdir,
+            arm_suffix=arm_suffix,
         )
         _copy_verifier(task_dir)
         custom_grader = _copy_custom_grader(task_dir, skill_path, grading_mode, evals_dir=evals_dir)
@@ -5446,6 +5455,7 @@ def generate_harbor_tasks(
     agent_workdir: str | None = None,
     evaluator_skill_path: Path | None = None,
     _baseline_alias_validation: _BaselineAliasValidation | None = None,
+    arm_suffix: str = "",
 ) -> list[Path]:
     """Generate tasks from one private evals snapshot, then publish exactly."""
 
@@ -5472,6 +5482,7 @@ def generate_harbor_tasks(
                 agent_workdir=agent_workdir,
                 evaluator_skill_path=private_skill_path,
                 _baseline_alias_validation=_baseline_alias_validation,
+                arm_suffix=arm_suffix,
             )
     if find_evals_file(evaluator_skill_path) is None:
         raise FileNotFoundError(f"No evals dataset found in {evaluator_skill_path / 'evals'}")
@@ -5540,6 +5551,7 @@ def generate_harbor_tasks(
             task_resources=task_resources,
             agent_workdir=agent_workdir,
             baseline_aliases_prevalidated=baseline_aliases_prevalidated,
+            arm_suffix=arm_suffix,
         )
         relative_tasks = [task.relative_to(private_output) for task in private_tasks]
         if output_requires_provenance:
diff --git a/src/skillevaluator/tier3/harbor/collector.py b/src/skillevaluator/tier3/harbor/collector.py
index a5a6985e..76c4e9a3 100644
--- a/src/skillevaluator/tier3/harbor/collector.py
+++ b/src/skillevaluator/tier3/harbor/collector.py
@@ -1848,6 +1848,11 @@ def _strip_attempt_suffix(value: str) -> str:
     return re.sub(r"(?:[-_])attempt\d+$", "", value)
 
 
+def _strip_arm_suffix(value: str) -> str:
+    """Remove SkillEvaluator dual-arm suffixes (-with-skill, -without-skill, or shorthand) from an identifier."""
+    return re.sub(r"-(?:with|without)(?:-skill)?$", "", value)
+
+
 def _canonical_case_id(value: str, expected_case_ids: set[str] | None = None) -> str:
     value = str(value or "").strip()
     if not value:
@@ -1858,9 +1863,12 @@ def _canonical_case_id(value: str, expected_case_ids: set[str] | None = None) ->
     if expected_case_ids and stripped in expected_case_ids:
         return stripped
     generated_prefix_stripped = stripped.removeprefix("skillevaluator-")
-    if expected_case_ids and generated_prefix_stripped in expected_case_ids:
-        return generated_prefix_stripped
-    return stripped
+    arm_stripped = _strip_arm_suffix(generated_prefix_stripped)
+    if expected_case_ids and arm_stripped in expected_case_ids:
+        return arm_stripped
+    if expected_case_ids:
+        return stripped
+    return arm_stripped
 
 
 def _entry_id(reward: dict[str, Any], expected_case_ids: set[str] | None = None) -> str:
diff --git a/src/skillevaluator/tier3/harbor/runner.py b/src/skillevaluator/tier3/harbor/runner.py
index ff6d1b7d..99a8c7c2 100644
--- a/src/skillevaluator/tier3/harbor/runner.py
+++ b/src/skillevaluator/tier3/harbor/runner.py
@@ -2290,6 +2290,9 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
     )
     staging_failure_stage = "with-skill-tasks"
     try:
+        is_dual_arm = not skip_baseline
+        with_arm_suffix = "-with-skill" if is_dual_arm else ""
+        without_arm_suffix = "-without-skill" if is_dual_arm else ""
         for agent in agents:
             with_dir = tasks_dir / agent / "with"
             without_dir = None if skip_baseline else tasks_dir / agent / "without"
@@ -2311,6 +2314,7 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
                 task_resources=resource_config,
                 agent_workdir=harbor_config.get("agent_workdir"),
                 evaluator_skill_path=evaluator_skill_path,
+                arm_suffix=with_arm_suffix,
             )
             task_names = [task.name for task in task_paths]
             if expected_task_names is None:
@@ -2352,6 +2356,7 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
                     agent_workdir=harbor_config.get("agent_workdir"),
                     evaluator_skill_path=evaluator_skill_path,
                     _baseline_alias_validation=baseline_alias_validation,
+                    arm_suffix=without_arm_suffix,
                 )
         if not skip_baseline:
             reporter.emit(ProgressEvent(stage="baseline-tasks", state="ready", detail="baseline inputs staged"))
diff --git a/tests/test_tier3_public_runtime.py b/tests/test_tier3_public_runtime.py
index a0135446..2a88f790 100644
--- a/tests/test_tier3_public_runtime.py
+++ b/tests/test_tier3_public_runtime.py
@@ -1281,3 +1281,63 @@ def handle_request(request: httpx.Request) -> httpx.Response:
 
     assert sdk_urls == [expected_url]
     assert verifier._anthropic_url() == expected_url
+
+
+def test_write_task_toml_dual_arm_suffix(tmp_path: Path) -> None:
+    """Verify that arm suffix is appended only when provided for dual-arm runs."""
+    case_dir = tmp_path / "case"
+    case_dir.mkdir()
+
+    # Default: no suffix (single-arm / standalone case)
+    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=True)
+    task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
+    assert task["task"]["name"] == "nvidia/skillevaluator-case-001"
+
+    # Dual-arm with-skill
+    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=True, arm_suffix="-with-skill")
+    task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
+    assert task["task"]["name"] == "nvidia/skillevaluator-case-001-with-skill"
+
+    # Dual-arm without-skill (baseline)
+    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=False, arm_suffix="-without-skill")
+    task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
+    assert task["task"]["name"] == "nvidia/skillevaluator-case-001-without-skill"
+
+
+def test_canonical_case_id_arm_stripping() -> None:
+    """Verify that _canonical_case_id strips arm suffixes whether expected_case_ids is supplied or not."""
+    from skillevaluator.tier3.harbor.collector import _canonical_case_id, _strip_arm_suffix
+
+    # Direct helper behavior
+    assert _strip_arm_suffix("case-001-with-skill") == "case-001"
+    assert _strip_arm_suffix("case-001-without-skill") == "case-001"
+    assert _strip_arm_suffix("case-001-with") == "case-001"
+    assert _strip_arm_suffix("case-001-without") == "case-001"
+    assert _strip_arm_suffix("case-001") == "case-001"
+
+    # Clean ID without suffix
+    assert _canonical_case_id("case-001") == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001") == "case-001"
+
+    # With expected_case_ids matching: canonical -with-skill / -without-skill
+    expected = {"case-001", "case-002"}
+    assert _canonical_case_id("case-001-with-skill", expected) == "case-001"
+    assert _canonical_case_id("case-001-without-skill", expected) == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-with-skill", expected) == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-without-skill", expected) == "case-001"
+
+    # With expected_case_ids matching: shorthand -with / -without
+    assert _canonical_case_id("case-001-with", expected) == "case-001"
+    assert _canonical_case_id("case-001-without", expected) == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-with", expected) == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-without", expected) == "case-001"
+
+    # Without expected_case_ids (fallback)
+    assert _canonical_case_id("case-001-with-skill") == "case-001"
+    assert _canonical_case_id("case-001-without-skill") == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-with-skill") == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-without-skill") == "case-001"
+    assert _canonical_case_id("case-001-with") == "case-001"
+    assert _canonical_case_id("case-001-without") == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-with") == "case-001"
+    assert _canonical_case_id("skillevaluator-case-001-without") == "case-001"

From 3b054db7e85d06637ca5725fb7f2f6decb088fe3 Mon Sep 17 00:00:00 2001
From: Karl Weinmeister <kweinmeister@google.com>
Date: Sun, 13 Sep 2026 14:09:05 -0500
Subject: [PATCH 2/4] fix(harbor): enhance dual-arm task naming and canonical
 case ID resolution

- Propagate arm suffixes to `[task] name` in staged native `task.toml` files.
- Decouple arm-suffix and attempt-suffix stripping with a commutative pipeline.
- Strip external repository and namespace prefixes before canonicalizing IDs.
- Protect case IDs retaining `skillevaluator-` or ending in `-with`/`-without`.
- Enforce fail-fast runtime string validation for `arm_suffix` across adapter APIs.
- Parameterize test suite and expand test matrix to cover all suffix variants.
- Document dual-arm evaluation fixes in CHANGELOG.md.

Signed-off-by: Karl Weinmeister <kweinmeister@google.com>
---
 CHANGELOG.md                                 |   5 +
 src/skillevaluator/tier3/harbor/adapter.py   |  40 ++-
 src/skillevaluator/tier3/harbor/collector.py |  42 ++-
 tests/test_tier3_public_runtime.py           | 295 +++++++++++++++----
 4 files changed, 316 insertions(+), 66 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0fc1c927..9d8544fd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -19,6 +19,11 @@ All notable changes to SkillEvaluator are documented in this file.
 
 ### Fixed
 
+- Tier 3 Harbor dual-arm evaluation propagates arm suffixes (`-with-skill`,
+  `-without-skill`) to `[task] name` in staged native `task.toml` files,
+  normalizes external repository and namespace prefixes, and commutatively
+  resolves canonical case IDs across attempt and arm suffix combinations
+  while preserving expected case IDs.
 - Malformed, non-UTF-8, or unreadable bundled and custom policy files now
   produce path-specific CLI errors instead of leaking raw parser or I/O errors
   ([#128](https://github.com/NVIDIA/SkillEvaluator/issues/128)).
diff --git a/src/skillevaluator/tier3/harbor/adapter.py b/src/skillevaluator/tier3/harbor/adapter.py
index d7265625..c0a0b1de 100644
--- a/src/skillevaluator/tier3/harbor/adapter.py
+++ b/src/skillevaluator/tier3/harbor/adapter.py
@@ -1804,6 +1804,8 @@ def _write_task_toml(
         raise TypeError("expected_skill must be a string before Harbor TOML serialization")
     if not isinstance(docker_image, str):
         raise TypeError("docker_image must be a string before Harbor TOML serialization")
+    if not isinstance(arm_suffix, str):
+        raise TypeError("arm_suffix must be a string before Harbor TOML serialization")
     docker_image_line = f"docker_image = {_toml_quote(docker_image)}\n" if docker_image else ""
     cpus = _task_resource_value(task_resources, "cpus", 2)
     memory_mb = _task_resource_value(task_resources, "memory_mb", 4096)
@@ -4227,6 +4229,32 @@ def _native_entry_id(task_dir: Path) -> str:
     return task_dir.name
 
 
+def _append_native_task_name_suffix(task_dir: Path, arm_suffix: str) -> None:
+    """Append dual-arm suffix to [task] name in a native task's task.toml."""
+    if not arm_suffix:
+        return
+    task_toml = task_dir / "task.toml"
+    if not task_toml.exists():
+        return
+    content = task_toml.read_text(encoding="utf-8")
+
+    pattern = r'(?ms)(\[task\]\s*?\n(?:(?!\[)[^\n]*\n)*?\s*name\s*=\s*)(["\'])(.*?)\2'
+
+    def _repl(m: re.Match[str]) -> str:
+        prefix, quote, old_name = m.groups()
+        if old_name.endswith(arm_suffix):
+            return m.group(0)
+        return f"{prefix}{quote}{old_name}{arm_suffix}{quote}"
+
+    new_content, count = re.subn(pattern, _repl, content, count=1)
+    if count == 0:
+        fallback_pattern = r'(?m)^(\s*name\s*=\s*)(["\'])(.*?)\2'
+        new_content = re.sub(fallback_pattern, _repl, content, count=1)
+
+    if new_content != content:
+        task_toml.write_text(new_content, encoding="utf-8")
+
+
 def _environment_reference_names(value: object) -> set[str]:
     """Return portable shell-style environment references from a TOML value."""
     if not isinstance(value, str):
@@ -4588,6 +4616,8 @@ def _stage_native_harbor_tasks_into(
     The source tree is copied first and all SkillEvaluator injections happen only in the
     staged result directory.
     """
+    if not isinstance(arm_suffix, str):
+        raise TypeError("arm_suffix must be a string before staging native Harbor tasks")
     _validate_runtime_discovery_env(runtime_env)
     _validate_runtime_loader_env(runtime_env)
     evals_dir = evaluator_skill_path / "evals"
@@ -4611,7 +4641,6 @@ def _stage_native_harbor_tasks_into(
         _native_task_workdir(source_task_dir)
     _ = task_resources
     _ = agent_workdir
-    _ = arm_suffix
 
     if output_dir.exists():
         shutil.rmtree(output_dir)
@@ -4640,6 +4669,7 @@ def _stage_native_harbor_tasks_into(
         baseline_aliases_prevalidated = True
     for task_dir in task_dirs:
         entry_id = _native_entry_id(task_dir)
+        _append_native_task_name_suffix(task_dir, arm_suffix)
         native_agent_workdir = _native_task_workdir(task_dir)
         _ensure_native_skills_dir(task_dir)
         entry = entries_by_id.get(entry_id)
@@ -4775,6 +4805,9 @@ def stage_native_harbor_tasks(
 ) -> list[Path]:
     """Stage native tasks privately, then publish one exact output snapshot."""
 
+    if not isinstance(arm_suffix, str):
+        raise TypeError("arm_suffix must be a string before staging native Harbor tasks")
+
     if evaluator_skill_path is None:
         with private_evaluator_skill_snapshot(skill_path, task_source="native_harbor") as private_skill_path:
             return stage_native_harbor_tasks(
@@ -4955,6 +4988,8 @@ def _generate_harbor_tasks_into(
     Returns:
         List of generated task directory paths.
     """
+    if not isinstance(arm_suffix, str):
+        raise TypeError("arm_suffix must be a string before generating Harbor tasks")
     _validate_runtime_discovery_env(runtime_env)
     _validate_runtime_loader_env(runtime_env)
     agent_workdir = _validated_agent_workdir(agent_workdir)
@@ -5459,6 +5494,9 @@ def generate_harbor_tasks(
 ) -> list[Path]:
     """Generate tasks from one private evals snapshot, then publish exactly."""
 
+    if not isinstance(arm_suffix, str):
+        raise TypeError("arm_suffix must be a string before generating Harbor tasks")
+
     if evaluator_skill_path is None:
         if find_evals_file(skill_path) is None:
             raise FileNotFoundError(f"No evals dataset found in {skill_path / 'evals'}")
diff --git a/src/skillevaluator/tier3/harbor/collector.py b/src/skillevaluator/tier3/harbor/collector.py
index 76c4e9a3..52ad9451 100644
--- a/src/skillevaluator/tier3/harbor/collector.py
+++ b/src/skillevaluator/tier3/harbor/collector.py
@@ -1849,26 +1849,42 @@ def _strip_attempt_suffix(value: str) -> str:
 
 
 def _strip_arm_suffix(value: str) -> str:
-    """Remove SkillEvaluator dual-arm suffixes (-with-skill, -without-skill, or shorthand) from an identifier."""
-    return re.sub(r"-(?:with|without)(?:-skill)?$", "", value)
+    """Remove SkillEvaluator dual-arm suffixes (-with-skill, -without-skill) from an identifier."""
+    return re.sub(r"-(?:with|without)-skill$", "", value)
+
+
+def _strip_arm_and_attempt_suffixes(value: str) -> str:
+    """Remove dual-arm and attempt suffixes regardless of ordering."""
+    pass1 = _strip_arm_suffix(_strip_attempt_suffix(value))
+    return _strip_attempt_suffix(_strip_arm_suffix(pass1))
 
 
 def _canonical_case_id(value: str, expected_case_ids: set[str] | None = None) -> str:
+    """Normalize a task, trial, or entry identifier to its canonical case ID."""
     value = str(value or "").strip()
     if not value:
         return ""
-    if expected_case_ids and value in expected_case_ids:
-        return value
-    stripped = _strip_attempt_suffix(value)
-    if expected_case_ids and stripped in expected_case_ids:
-        return stripped
-    generated_prefix_stripped = stripped.removeprefix("skillevaluator-")
-    arm_stripped = _strip_arm_suffix(generated_prefix_stripped)
-    if expected_case_ids and arm_stripped in expected_case_ids:
-        return arm_stripped
+    name_part = value.rpartition("/")[2] or value
+
+    raw_candidates = [
+        name_part,
+        _strip_attempt_suffix(name_part),
+        _strip_arm_suffix(name_part),
+        _strip_arm_and_attempt_suffixes(name_part),
+    ]
+    candidates: list[str] = []
+    seen: set[str] = set()
+    for candidate in raw_candidates + [c.removeprefix("skillevaluator-") for c in raw_candidates]:
+        if candidate and candidate not in seen:
+            seen.add(candidate)
+            candidates.append(candidate)
+
     if expected_case_ids:
-        return stripped
-    return arm_stripped
+        for candidate in candidates:
+            if candidate in expected_case_ids:
+                return candidate
+
+    return _strip_arm_and_attempt_suffixes(name_part.removeprefix("skillevaluator-"))
 
 
 def _entry_id(reward: dict[str, Any], expected_case_ids: set[str] | None = None) -> str:
diff --git a/tests/test_tier3_public_runtime.py b/tests/test_tier3_public_runtime.py
index 2a88f790..3a02bc67 100644
--- a/tests/test_tier3_public_runtime.py
+++ b/tests/test_tier3_public_runtime.py
@@ -24,7 +24,14 @@
 from skillevaluator.provider_config import ProviderConfig, resolve_llm_provider
 from skillevaluator.tier3 import commands as tier3_commands
 from skillevaluator.tier3.evals_config import EvalsConfigError, load_evals_config
-from skillevaluator.tier3.harbor.adapter import _EVALUATOR_MANAGED_RUNTIME_ENV, _write_task_toml
+from skillevaluator.tier3.harbor.adapter import (
+    _EVALUATOR_MANAGED_RUNTIME_ENV,
+    _generate_harbor_tasks_into,
+    _stage_native_harbor_tasks_into,
+    _write_task_toml,
+    generate_harbor_tasks,
+    stage_native_harbor_tasks,
+)
 from skillevaluator.tier3.harbor.runner import (
     _model_for_agent,
     _nvidia_build_agent_import_path,
@@ -1283,61 +1290,245 @@ def handle_request(request: httpx.Request) -> httpx.Response:
     assert verifier._anthropic_url() == expected_url
 
 
-def test_write_task_toml_dual_arm_suffix(tmp_path: Path) -> None:
+@pytest.mark.parametrize(
+    ("has_skill", "arm_suffix", "expected_task_name"),
+    [
+        (True, "", "nvidia/skillevaluator-case-001"),
+        (True, "-with-skill", "nvidia/skillevaluator-case-001-with-skill"),
+        (False, "-without-skill", "nvidia/skillevaluator-case-001-without-skill"),
+    ],
+)
+def test_write_task_toml_dual_arm_suffix(
+    tmp_path: Path,
+    has_skill: bool,
+    arm_suffix: str,
+    expected_task_name: str,
+) -> None:
     """Verify that arm suffix is appended only when provided for dual-arm runs."""
     case_dir = tmp_path / "case"
     case_dir.mkdir()
-
-    # Default: no suffix (single-arm / standalone case)
-    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=True)
+    _write_task_toml(
+        case_dir,
+        {"id": "case-001", "expected_skill": "demo"},
+        has_skill=has_skill,
+        arm_suffix=arm_suffix,
+    )
     task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
-    assert task["task"]["name"] == "nvidia/skillevaluator-case-001"
+    assert task["task"]["name"] == expected_task_name
 
-    # Dual-arm with-skill
-    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=True, arm_suffix="-with-skill")
-    task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
+
+def test_write_task_toml_type_safety(tmp_path: Path) -> None:
+    """Verify that _write_task_toml rejects non-string arm_suffix values."""
+    case_dir = tmp_path / "case"
+    case_dir.mkdir()
+
+    with pytest.raises(TypeError, match="arm_suffix must be a string"):
+        _write_task_toml(case_dir, {"id": "case-001"}, has_skill=True, arm_suffix=123)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("case-001-with-skill", "case-001"),
+        ("case-001-without-skill", "case-001"),
+        ("case-001-with", "case-001-with"),
+        ("case-001-without", "case-001-without"),
+        ("case-001", "case-001"),
+    ],
+)
+def test_strip_arm_suffix(value: str, expected: str) -> None:
+    """Verify _strip_arm_suffix trims exact dual-arm suffixes."""
+    from skillevaluator.tier3.harbor.collector import _strip_arm_suffix
+
+    assert _strip_arm_suffix(value) == expected
+
+
+@pytest.mark.parametrize(
+    ("value", "expected"),
+    [
+        ("case-001-with-skill-attempt1", "case-001"),
+        ("case-001-without-skill_attempt2", "case-001"),
+        ("case-001-attempt1-with-skill", "case-001"),
+        ("case-001_attempt2-without-skill", "case-001"),
+        ("case-001-with-skill", "case-001"),
+        ("case-001-without-skill", "case-001"),
+        ("case-001-attempt1", "case-001"),
+        ("case-001", "case-001"),
+    ],
+)
+def test_strip_arm_and_attempt_suffixes(value: str, expected: str) -> None:
+    """Verify _strip_arm_and_attempt_suffixes removes both suffixes regardless of ordering."""
+    from skillevaluator.tier3.harbor.collector import _strip_arm_and_attempt_suffixes
+
+    assert _strip_arm_and_attempt_suffixes(value) == expected
+
+
+@pytest.mark.parametrize(
+    ("raw_input", "expected_ids", "expected_output"),
+    [
+        # Bare case IDs without suffixes
+        ("case-001", None, "case-001"),
+        ("skillevaluator-case-001", None, "case-001"),
+        ("skillevaluator-case-001", {"case-001", "case-002"}, "case-001"),
+        # Dual-arm suffixes without expected_case_ids (fallback)
+        ("case-001-with-skill", None, "case-001"),
+        ("case-001-without-skill", None, "case-001"),
+        ("skillevaluator-case-001-with-skill", None, "case-001"),
+        ("skillevaluator-case-001-without-skill", None, "case-001"),
+        # Dual-arm suffixes with expected_case_ids matching
+        ("case-001-with-skill", {"case-001", "case-002"}, "case-001"),
+        ("case-001-without-skill", {"case-001", "case-002"}, "case-001"),
+        ("skillevaluator-case-001-with-skill", {"case-001", "case-002"}, "case-001"),
+        ("skillevaluator-case-001-without-skill", {"case-001", "case-002"}, "case-001"),
+        # Namespaced task names (e.g., nvidia/...)
+        ("nvidia/skillevaluator-case-001", None, "case-001"),
+        ("nvidia/skillevaluator-case-001", {"case-001"}, "case-001"),
+        ("nvidia/skillevaluator-case-001-with-skill", None, "case-001"),
+        ("nvidia/skillevaluator-case-001-with-skill", {"case-001"}, "case-001"),
+        ("nvidia/skillevaluator-case-001-without-skill", None, "case-001"),
+        ("nvidia/skillevaluator-case-001-without-skill", {"case-001"}, "case-001"),
+        ("custom/repo/skillevaluator-case-002-with-skill", {"case-002"}, "case-002"),
+        # Attempt suffixes combined with dual-arm suffixes in both orderings
+        ("case-001-with-skill-attempt1", None, "case-001"),
+        ("case-001-with-skill-attempt1", {"case-001"}, "case-001"),
+        ("case-001-without-skill_attempt2", None, "case-001"),
+        ("case-001-without-skill_attempt2", {"case-001"}, "case-001"),
+        ("case-001-attempt1-with-skill", None, "case-001"),
+        ("case-001-attempt1-with-skill", {"case-001"}, "case-001"),
+        ("case-001_attempt2-without-skill", None, "case-001"),
+        ("case-001_attempt2-without-skill", {"case-001"}, "case-001"),
+        ("skillevaluator-case-001-with-skill-attempt1", None, "case-001"),
+        ("skillevaluator-case-001-with-skill-attempt1", {"case-001"}, "case-001"),
+        ("skillevaluator-case-001-attempt1-with-skill", None, "case-001"),
+        ("skillevaluator-case-001-attempt1-with-skill", {"case-001"}, "case-001"),
+        ("nvidia/skillevaluator-case-001-with-skill-attempt3", {"case-001"}, "case-001"),
+        ("nvidia/skillevaluator-case-001-attempt3-with-skill", {"case-001"}, "case-001"),
+        ("nvidia/skillevaluator-case-001-without-skill_attempt4", None, "case-001"),
+        ("nvidia/skillevaluator-case-001_attempt4-without-skill", None, "case-001"),
+        # Legitimate case IDs ending with -with or -without preserved when in expected_case_ids
+        ("case-with", {"case-with"}, "case-with"),
+        ("case-without", {"case-without"}, "case-without"),
+        ("nvidia/case-with", {"case-with"}, "case-with"),
+        ("nvidia/case-without", {"case-without"}, "case-without"),
+        ("skillevaluator-case-with", {"case-with"}, "case-with"),
+        ("skillevaluator-case-without", {"case-without"}, "case-without"),
+        ("case-with-attempt1", {"case-with"}, "case-with"),
+        ("case-without-attempt2", {"case-without"}, "case-without"),
+        # Dual-arm runs on legitimate -with / -without IDs
+        ("case-with-with-skill", {"case-with"}, "case-with"),
+        ("case-without-without-skill", {"case-without"}, "case-without"),
+        ("skillevaluator-case-with-with-skill", {"case-with"}, "case-with"),
+        ("skillevaluator-case-without-without-skill", {"case-without"}, "case-without"),
+        ("nvidia/skillevaluator-case-with-with-skill", {"case-with"}, "case-with"),
+        ("nvidia/skillevaluator-case-without-without-skill", {"case-without"}, "case-without"),
+        ("case-with-with-skill-attempt1", {"case-with"}, "case-with"),
+        ("case-with-attempt1-with-skill", {"case-with"}, "case-with"),
+        ("case-without-without-skill_attempt2", {"case-without"}, "case-without"),
+        ("case-without_attempt2-without-skill", {"case-without"}, "case-without"),
+        # Case IDs that retain the skillevaluator- prefix in expected_case_ids
+        ("skillevaluator-case-001-with-skill", {"skillevaluator-case-001"}, "skillevaluator-case-001"),
+        ("skillevaluator-case-001-without-skill", {"skillevaluator-case-001"}, "skillevaluator-case-001"),
+        ("nvidia/skillevaluator-case-001-with-skill", {"skillevaluator-case-001"}, "skillevaluator-case-001"),
+        ("skillevaluator-case-001-with-skill-attempt1", {"skillevaluator-case-001"}, "skillevaluator-case-001"),
+        ("skillevaluator-case-001-attempt1-with-skill", {"skillevaluator-case-001"}, "skillevaluator-case-001"),
+        # Empty and blank strings
+        ("", None, ""),
+        ("   ", None, ""),
+        ("", {"case-001"}, ""),
+    ],
+)
+def test_canonical_case_id_arm_stripping(
+    raw_input: str,
+    expected_ids: set[str] | None,
+    expected_output: str,
+) -> None:
+    """Verify _canonical_case_id normalizes task identifiers across naming and attempt variants."""
+    from skillevaluator.tier3.harbor.collector import _canonical_case_id
+
+    assert _canonical_case_id(raw_input, expected_ids) == expected_output
+
+
+@pytest.mark.parametrize(
+    ("arm_suffix", "expected_task_name"),
+    [
+        ("-with-skill", "nvidia/case-001-with-skill"),
+        ("-without-skill", "nvidia/case-001-without-skill"),
+    ],
+)
+def test_stage_native_harbor_tasks_dual_arm_suffix(
+    tmp_path: Path,
+    arm_suffix: str,
+    expected_task_name: str,
+) -> None:
+    """Verify that stage_native_harbor_tasks appends arm suffix to native task.toml name."""
+    skill_dir = tmp_path / "target-skill"
+    skill_dir.mkdir()
+    (skill_dir / "SKILL.md").write_text("# Target Skill\n", encoding="utf-8")
+    evals_dir = skill_dir / "evals" / "harbor"
+    task_dir = evals_dir / "case-001"
+    task_dir.mkdir(parents=True)
+    (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
+    (task_dir / "task.toml").write_text(
+        'schema_version = "1.3"\n\n[task]\nname = "nvidia/case-001"\n\n[environment]\n',
+        encoding="utf-8",
+    )
+    tests_dir = task_dir / "tests"
+    tests_dir.mkdir()
+    (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+    out_dir = tmp_path / f"out{arm_suffix}"
+    staged = stage_native_harbor_tasks(
+        skill_dir,
+        out_dir,
+        grading_mode="custom_only",
+        arm_suffix=arm_suffix,
+    )[0]
+    task = tomllib.loads((staged / "task.toml").read_text(encoding="utf-8"))
+    assert task["task"]["name"] == expected_task_name
+
+
+def test_stage_native_harbor_tasks_type_safety(tmp_path: Path) -> None:
+    """Verify that native staging functions reject non-string arm_suffix values."""
+    skill_dir = tmp_path / "target-skill"
+    skill_dir.mkdir()
+
+    with pytest.raises(TypeError, match="arm_suffix must be a string"):
+        stage_native_harbor_tasks(skill_dir, tmp_path / "err", arm_suffix=123)  # type: ignore[arg-type]
+
+    with pytest.raises(TypeError, match="arm_suffix must be a string"):
+        _stage_native_harbor_tasks_into(
+            skill_dir,
+            tmp_path / "err",
+            evaluator_skill_path=skill_dir,
+            arm_suffix=123,  # type: ignore[arg-type]
+        )
+
+
+def test_generate_harbor_tasks_dual_arm_suffix(tmp_path: Path) -> None:
+    """Verify that generate_harbor_tasks propagates arm suffix into task.toml name."""
+    skill_dir = tmp_path / "gen-skill"
+    skill_dir.mkdir()
+    (skill_dir / "SKILL.md").write_text("# Gen Skill\n", encoding="utf-8")
+    evals_dir = skill_dir / "evals"
+    evals_dir.mkdir()
+    (evals_dir / "evals.json").write_text(
+        json.dumps([{"id": "case-001", "prompt": "test prompt"}]),
+        encoding="utf-8",
+    )
+
+    out = tmp_path / "gen_out"
+    staged = generate_harbor_tasks(skill_dir, out, arm_suffix="-with-skill")[0]
+    task = tomllib.loads((staged / "task.toml").read_text(encoding="utf-8"))
     assert task["task"]["name"] == "nvidia/skillevaluator-case-001-with-skill"
 
-    # Dual-arm without-skill (baseline)
-    _write_task_toml(case_dir, {"id": "case-001", "expected_skill": "demo"}, has_skill=False, arm_suffix="-without-skill")
-    task = tomllib.loads((case_dir / "task.toml").read_text(encoding="utf-8"))
-    assert task["task"]["name"] == "nvidia/skillevaluator-case-001-without-skill"
-
-
-def test_canonical_case_id_arm_stripping() -> None:
-    """Verify that _canonical_case_id strips arm suffixes whether expected_case_ids is supplied or not."""
-    from skillevaluator.tier3.harbor.collector import _canonical_case_id, _strip_arm_suffix
-
-    # Direct helper behavior
-    assert _strip_arm_suffix("case-001-with-skill") == "case-001"
-    assert _strip_arm_suffix("case-001-without-skill") == "case-001"
-    assert _strip_arm_suffix("case-001-with") == "case-001"
-    assert _strip_arm_suffix("case-001-without") == "case-001"
-    assert _strip_arm_suffix("case-001") == "case-001"
-
-    # Clean ID without suffix
-    assert _canonical_case_id("case-001") == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001") == "case-001"
-
-    # With expected_case_ids matching: canonical -with-skill / -without-skill
-    expected = {"case-001", "case-002"}
-    assert _canonical_case_id("case-001-with-skill", expected) == "case-001"
-    assert _canonical_case_id("case-001-without-skill", expected) == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-with-skill", expected) == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-without-skill", expected) == "case-001"
-
-    # With expected_case_ids matching: shorthand -with / -without
-    assert _canonical_case_id("case-001-with", expected) == "case-001"
-    assert _canonical_case_id("case-001-without", expected) == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-with", expected) == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-without", expected) == "case-001"
-
-    # Without expected_case_ids (fallback)
-    assert _canonical_case_id("case-001-with-skill") == "case-001"
-    assert _canonical_case_id("case-001-without-skill") == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-with-skill") == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-without-skill") == "case-001"
-    assert _canonical_case_id("case-001-with") == "case-001"
-    assert _canonical_case_id("case-001-without") == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-with") == "case-001"
-    assert _canonical_case_id("skillevaluator-case-001-without") == "case-001"
+    with pytest.raises(TypeError, match="arm_suffix must be a string"):
+        generate_harbor_tasks(skill_dir, tmp_path / "err", arm_suffix=123)  # type: ignore[arg-type]
+
+    with pytest.raises(TypeError, match="arm_suffix must be a string"):
+        _generate_harbor_tasks_into(
+            skill_dir,
+            tmp_path / "err",
+            evaluator_skill_path=skill_dir,
+            arm_suffix=123,  # type: ignore[arg-type]
+        )
+

From 18cac4a7252f40905850dd5377a6a25470291b21 Mon Sep 17 00:00:00 2001
From: Karl Weinmeister <kweinmeister@google.com>
Date: Sun, 27 Sep 2026 09:51:37 -0500
Subject: [PATCH 3/4] fix(harbor): preserve authored task IDs and update native
 task.toml structurally

Signed-off-by: Karl Weinmeister <kweinmeister@google.com>
---
 src/skillevaluator/tier3/harbor/adapter.py    | 238 ++++++++++++++-
 src/skillevaluator/tier3/harbor/collector.py  | 177 +++++++++---
 tests/test_tier3_public_runtime.py            | 271 ++++++++++++++++--
 .../tier3/test_harbor_security_attribution.py |  22 ++
 4 files changed, 635 insertions(+), 73 deletions(-)

diff --git a/src/skillevaluator/tier3/harbor/adapter.py b/src/skillevaluator/tier3/harbor/adapter.py
index c0a0b1de..974ef997 100644
--- a/src/skillevaluator/tier3/harbor/adapter.py
+++ b/src/skillevaluator/tier3/harbor/adapter.py
@@ -4229,28 +4229,240 @@ def _native_entry_id(task_dir: Path) -> str:
     return task_dir.name
 
 
-def _append_native_task_name_suffix(task_dir: Path, arm_suffix: str) -> None:
+_TOML_DOUBLE_QUOTE = '"'
+_TOML_SINGLE_QUOTE = "'"
+_TOML_MULTILINE_DOUBLE_QUOTE = '"""'
+_TOML_MULTILINE_SINGLE_QUOTE = "'''"
+_TOML_STRING_DELIMITERS = (
+    _TOML_MULTILINE_DOUBLE_QUOTE,
+    _TOML_MULTILINE_SINGLE_QUOTE,
+    _TOML_DOUBLE_QUOTE,
+    _TOML_SINGLE_QUOTE,
+)
+_TOML_ESCAPE_PAIR_LEN = 2  # backslash + escaped character
+
+
+def _skip_toml_string_literal(content: str, start: int) -> int:
+    """Return index immediately after the TOML string literal starting at *start*."""
+    n = len(content)
+    for delimiter in _TOML_STRING_DELIMITERS:
+        if not content.startswith(delimiter, start):
+            continue
+        delim_len = len(delimiter)
+        quote_char = delimiter[0]
+        supports_escapes = quote_char == _TOML_DOUBLE_QUOTE
+        is_multiline = delim_len > 1
+
+        i = start + delim_len
+        while i < n:
+            if supports_escapes and content[i] == "\\":
+                i += _TOML_ESCAPE_PAIR_LEN
+                continue
+            if content.startswith(delimiter, i):
+                i += delim_len
+                if is_multiline:
+                    while i < n and content[i] == quote_char:
+                        i += 1
+                return i
+            i += 1
+        return n
+    return start
+
+
+def _parse_toml_dotted_key(raw: str) -> tuple[str, ...]:
+    """Parse a TOML key or table path into unquoted segment names."""
+    parts: list[str] = []
+    for token in re.findall(r'"(?:\\.|[^"\\\r\n])*"|\'[^\'\r\n]*\'|[A-Za-z0-9_-]+', raw):
+        if (token.startswith('"') and token.endswith('"')) or (token.startswith("'") and token.endswith("'")):
+            parts.append(token[1:-1])
+        else:
+            parts.append(token)
+    return tuple(parts)
+
+
+def _find_toml_table_key_value_span(content: str, target_table: str, target_key: str) -> tuple[int, int] | None:
+    """Return the (start, end) character span of a string value for [target_table].target_key."""
+    n = len(content)
+    i = 0
+    current_table: tuple[str, ...] = ()
+    bracket_depth = 0
+
+    while i < n:
+        while i < n and content[i] in " \t":
+            i += 1
+        if i >= n:
+            break
+        if content[i] in "\r\n":
+            i += 1
+            continue
+        if content[i] == "#":
+            while i < n and content[i] != "\n":
+                i += 1
+            continue
+
+        if bracket_depth == 0 and content[i] == "[":
+            line_end = content.find("\n", i)
+            if line_end == -1:
+                line_end = n
+            line = content[i:line_end]
+            header_match = re.match(r"^\[(\[?)\s*([^\[\]#\r\n]+?)\s*\]\]?\s*(?:#.*)?\r?$", line)
+            if header_match:
+                is_array_table, raw_table = header_match.groups()
+                table_parts = _parse_toml_dotted_key(raw_table)
+                current_table = (f"[[{'.'.join(table_parts)}]]",) if is_array_table else table_parts
+                i = line_end + 1
+                continue
+
+        if bracket_depth == 0:
+            key_match = re.match(
+                r"""^(?:[A-Za-z0-9_-]+|"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*')(?:\s*\.\s*(?:[A-Za-z0-9_-]+|"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*'))*\s*=""",
+                content[i:],
+            )
+            if key_match:
+                raw_lhs = content[i : i + key_match.end() - 1].strip()
+                key_parts = _parse_toml_dotted_key(raw_lhs)
+                full_path = (*current_table, *key_parts)
+                val_start = i + key_match.end()
+                while val_start < n and content[val_start] in " \t":
+                    val_start += 1
+                if full_path == (target_table, target_key) and val_start < n and content[val_start] in "\"'":
+                    val_end = _skip_toml_string_literal(content, val_start)
+                    return val_start, val_end
+                i = val_start
+
+        while i < n:
+            ch = content[i]
+            if ch in "\"'":
+                i = _skip_toml_string_literal(content, i)
+                continue
+            if ch in "[({":
+                bracket_depth += 1
+                i += 1
+                continue
+            if ch in "])}":
+                bracket_depth = max(0, bracket_depth - 1)
+                i += 1
+                continue
+            if ch == "#":
+                while i < n and content[i] != "\n":
+                    i += 1
+                continue
+            if ch == "\n":
+                i += 1
+                if bracket_depth == 0:
+                    break
+                continue
+            i += 1
+
+    return None
+
+
+def _ensure_native_metadata_entry_id(content: str, entry_id: str) -> str:
+    """Ensure [metadata].entry_id is present in a native task.toml document."""
+    n = len(content)
+    i = 0
+    bracket_depth = 0
+    while i < n:
+        while i < n and content[i] in " \t":
+            i += 1
+        if i >= n:
+            break
+        if content[i] in "\r\n":
+            i += 1
+            continue
+        if content[i] == "#":
+            while i < n and content[i] != "\n":
+                i += 1
+            continue
+        if bracket_depth == 0 and content[i] == "[":
+            line_end = content.find("\n", i)
+            if line_end == -1:
+                line_end = n
+            line = content[i:line_end]
+            header_match = re.match(r"^\[(\[?)\s*([^\[\]#\r\n]+?)\s*\]\]?\s*(?:#.*)?\r?$", line)
+            if header_match:
+                is_array_table, raw_table = header_match.groups()
+                if not is_array_table and _parse_toml_dotted_key(raw_table) == ("metadata",):
+                    insert_pos = line_end + 1 if line_end < n else n
+                    prefix = content[:insert_pos]
+                    if not prefix.endswith("\n"):
+                        prefix += "\n"
+                    return f"{prefix}entry_id = {_toml_quote(entry_id)}\n{content[insert_pos:]}"
+                i = line_end + 1
+                continue
+        while i < n:
+            ch = content[i]
+            if ch in "\"'":
+                i = _skip_toml_string_literal(content, i)
+                continue
+            if ch in "[({":
+                bracket_depth += 1
+                i += 1
+                continue
+            if ch in "])}":
+                bracket_depth = max(0, bracket_depth - 1)
+                i += 1
+                continue
+            if ch == "#":
+                while i < n and content[i] != "\n":
+                    i += 1
+                continue
+            if ch == "\n":
+                i += 1
+                if bracket_depth == 0:
+                    break
+                continue
+            i += 1
+
+    suffix = "" if content.endswith("\n") else "\n"
+    return f"{content}{suffix}\n[metadata]\nentry_id = {_toml_quote(entry_id)}\n"
+
+
+def _append_native_task_name_suffix(
+    task_dir: Path,
+    arm_suffix: str,
+    *,
+    entry_id: str | None = None,
+) -> None:
     """Append dual-arm suffix to [task] name in a native task's task.toml."""
     if not arm_suffix:
         return
     task_toml = task_dir / "task.toml"
     if not task_toml.exists():
         return
-    content = task_toml.read_text(encoding="utf-8")
+    try:
+        content = task_toml.read_text(encoding="utf-8")
+        data = tomllib.loads(content)
+    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError):
+        return
 
-    pattern = r'(?ms)(\[task\]\s*?\n(?:(?!\[)[^\n]*\n)*?\s*name\s*=\s*)(["\'])(.*?)\2'
+    task_table = data.get("task")
+    if not isinstance(task_table, dict):
+        return
+    old_name = task_table.get("name")
+    if not isinstance(old_name, str):
+        return
 
-    def _repl(m: re.Match[str]) -> str:
-        prefix, quote, old_name = m.groups()
-        if old_name.endswith(arm_suffix):
-            return m.group(0)
-        return f"{prefix}{quote}{old_name}{arm_suffix}{quote}"
+    new_name = f"{old_name}{arm_suffix}"
+    span = _find_toml_table_key_value_span(content, "task", "name")
+    if span is None:
+        return
+    val_start, val_end = span
+    new_content = f"{content[:val_start]}{_toml_quote(new_name)}{content[val_end:]}"
 
-    new_content, count = re.subn(pattern, _repl, content, count=1)
-    if count == 0:
-        fallback_pattern = r'(?m)^(\s*name\s*=\s*)(["\'])(.*?)\2'
-        new_content = re.sub(fallback_pattern, _repl, content, count=1)
+    metadata = data.get("metadata")
+    effective_entry_id = entry_id or (
+        str(metadata["entry_id"]) if isinstance(metadata, dict) and metadata.get("entry_id") else task_dir.name
+    )
+    if effective_entry_id and not (isinstance(metadata, dict) and "entry_id" in metadata):
+        candidate_content = _ensure_native_metadata_entry_id(new_content, effective_entry_id)
+        try:
+            tomllib.loads(candidate_content)
+            new_content = candidate_content
+        except tomllib.TOMLDecodeError:
+            pass
 
+    tomllib.loads(new_content)
     if new_content != content:
         task_toml.write_text(new_content, encoding="utf-8")
 
@@ -4669,7 +4881,7 @@ def _stage_native_harbor_tasks_into(
         baseline_aliases_prevalidated = True
     for task_dir in task_dirs:
         entry_id = _native_entry_id(task_dir)
-        _append_native_task_name_suffix(task_dir, arm_suffix)
+        _append_native_task_name_suffix(task_dir, arm_suffix, entry_id=entry_id)
         native_agent_workdir = _native_task_workdir(task_dir)
         _ensure_native_skills_dir(task_dir)
         entry = entries_by_id.get(entry_id)
diff --git a/src/skillevaluator/tier3/harbor/collector.py b/src/skillevaluator/tier3/harbor/collector.py
index 52ad9451..e807182d 100644
--- a/src/skillevaluator/tier3/harbor/collector.py
+++ b/src/skillevaluator/tier3/harbor/collector.py
@@ -1577,7 +1577,7 @@ def _merge_constituent_default_reward_failure(
     )
 
 
-def _extract_rewards(job_dir: Path) -> list[dict[str, Any]]:
+def _extract_rewards(job_dir: Path, *, arm_suffix: str = "") -> list[dict[str, Any]]:
     """Extract reward.json from all trials in a job directory."""
     rewards: list[dict[str, Any]] = []
     scored_trial_roots: set[Path] = set()
@@ -1605,10 +1605,13 @@ def _extract_rewards(job_dir: Path) -> list[dict[str, Any]]:
         data["_trial_name"] = trial_name
         data["_trial_root_name"] = trial_dir.name
         data["_started_at"] = result.get("started_at")
+        if arm_suffix:
+            data["_arm_suffix"] = arm_suffix
         if not data.get("entry_id"):
-            entry_id = _entry_id_from_harbor_result(result)
+            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
             if entry_id:
                 data["entry_id"] = entry_id
+                data["_result_entry_id"] = entry_id
         traj_file = _reward_trajectory_path(trial_dir, None)
         if traj_file.exists():
             data["_has_trajectory"] = True
@@ -1637,6 +1640,8 @@ def _extract_rewards(job_dir: Path) -> list[dict[str, Any]]:
                     continue
                 data["_trial_name"] = trial_name
                 data["_trial_root_name"] = trial_dir.name
+                if arm_suffix:
+                    data["_arm_suffix"] = arm_suffix
                 if step_name:
                     data["_step_name"] = step_name
                 result_file = trial_dir / "result.json"
@@ -1646,9 +1651,10 @@ def _extract_rewards(job_dir: Path) -> list[dict[str, Any]]:
                         _merge_constituent_default_reward_failure(data, result, trial_dir)
                         data["_started_at"] = result.get("started_at")
                         if not data.get("entry_id"):
-                            entry_id = _entry_id_from_harbor_result(result)
+                            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
                             if entry_id:
                                 data["entry_id"] = entry_id
+                                data["_result_entry_id"] = entry_id
                 traj_file = _reward_trajectory_path(trial_dir, step_name)
                 if traj_file.exists():
                     data["_has_trajectory"] = True
@@ -1678,10 +1684,13 @@ def _extract_rewards(job_dir: Path) -> list[dict[str, Any]]:
         data["_trial_name"] = trial_name
         data["_trial_root_name"] = trial_dir.name
         data["_started_at"] = result.get("started_at")
+        if arm_suffix:
+            data["_arm_suffix"] = arm_suffix
         if not data.get("entry_id"):
-            entry_id = _entry_id_from_harbor_result(result)
+            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
             if entry_id:
                 data["entry_id"] = entry_id
+                data["_result_entry_id"] = entry_id
         traj_file = _reward_trajectory_path(trial_dir, None)
         if traj_file.exists():
             data["_has_trajectory"] = True
@@ -1765,26 +1774,53 @@ def _harbor_result_rewards(result: dict[str, Any]) -> dict[str, Any] | None:
     return aggregated or None
 
 
-def _entry_id_from_harbor_result(result: dict[str, Any]) -> str:
-    task_name = result.get("task_name")
-    if isinstance(task_name, str) and task_name.strip():
-        return task_name.strip().rsplit("/", 1)[-1]
+def _result_task_path_entry_id(task_path_value: Any) -> tuple[str, str]:
+    """Return (metadata_entry_id, dir_name) from a Harbor task path when available."""
+    if not isinstance(task_path_value, str) or not task_path_value.strip():
+        return "", ""
+    task_dir = Path(task_path_value.strip())
+    task_toml = task_dir / "task.toml"
+    if not task_toml.is_symlink() and task_toml.is_file():
+        raw_toml = _read_bounded_text(task_toml)
+        if raw_toml is not None:
+            try:
+                import tomllib
 
+                parsed = tomllib.loads(raw_toml)
+                metadata = parsed.get("metadata") if isinstance(parsed, dict) else None
+                if isinstance(metadata, dict) and metadata.get("entry_id"):
+                    return str(metadata["entry_id"]).strip(), task_dir.name
+            except Exception:
+                pass
+    return "", task_dir.name
+
+
+def _entry_id_from_harbor_result(result: dict[str, Any], *, arm_suffix: str = "") -> str:
+    fallback_dir_name = ""
     task_id = result.get("task_id")
     if isinstance(task_id, dict):
-        task_path = task_id.get("path")
-        if isinstance(task_path, str) and task_path.strip():
-            return Path(task_path).name
+        meta_id, dir_name = _result_task_path_entry_id(task_id.get("path"))
+        if meta_id:
+            return meta_id
+        fallback_dir_name = dir_name or fallback_dir_name
 
     config = result.get("config")
     if isinstance(config, dict):
         task = config.get("task")
         if isinstance(task, dict):
-            task_path = task.get("path")
-            if isinstance(task_path, str) and task_path.strip():
-                return Path(task_path).name
+            meta_id, dir_name = _result_task_path_entry_id(task.get("path"))
+            if meta_id:
+                return meta_id
+            fallback_dir_name = dir_name or fallback_dir_name
 
-    return ""
+    task_name = result.get("task_name")
+    if isinstance(task_name, str) and task_name.strip():
+        short_name = _strip_attempt_suffix(task_name.strip().rsplit("/", 1)[-1])
+        if arm_suffix and short_name.endswith(arm_suffix):
+            short_name = short_name.removesuffix(arm_suffix)
+        return _strip_attempt_suffix(short_name)
+
+    return fallback_dir_name
 
 
 def _overall_score(reward: dict[str, Any]) -> float | None:
@@ -1854,45 +1890,83 @@ def _strip_arm_suffix(value: str) -> str:
 
 
 def _strip_arm_and_attempt_suffixes(value: str) -> str:
-    """Remove dual-arm and attempt suffixes regardless of ordering."""
-    pass1 = _strip_arm_suffix(_strip_attempt_suffix(value))
-    return _strip_attempt_suffix(_strip_arm_suffix(pass1))
+    """Remove at most one dual-arm suffix and one attempt suffix regardless of ordering."""
+    attempt_stripped = _strip_attempt_suffix(value)
+    if attempt_stripped != value:
+        return _strip_arm_suffix(attempt_stripped)
+    return _strip_attempt_suffix(_strip_arm_suffix(value))
 
 
-def _canonical_case_id(value: str, expected_case_ids: set[str] | None = None) -> str:
+def _canonical_case_id(
+    value: str,
+    expected_case_ids: set[str] | None = None,
+    *,
+    strip_arm: bool = True,
+) -> str:
     """Normalize a task, trial, or entry identifier to its canonical case ID."""
     value = str(value or "").strip()
     if not value:
         return ""
+    if expected_case_ids and value in expected_case_ids:
+        return value
     name_part = value.rpartition("/")[2] or value
 
-    raw_candidates = [
-        name_part,
-        _strip_attempt_suffix(name_part),
-        _strip_arm_suffix(name_part),
-        _strip_arm_and_attempt_suffixes(name_part),
-    ]
-    candidates: list[str] = []
-    seen: set[str] = set()
-    for candidate in raw_candidates + [c.removeprefix("skillevaluator-") for c in raw_candidates]:
-        if candidate and candidate not in seen:
-            seen.add(candidate)
-            candidates.append(candidate)
-
     if expected_case_ids:
+        raw_candidates = [
+            name_part,
+            _strip_attempt_suffix(name_part),
+        ]
+        if strip_arm:
+            raw_candidates.extend(
+                [
+                    _strip_arm_suffix(name_part),
+                    _strip_arm_and_attempt_suffixes(name_part),
+                ]
+            )
+        candidates: list[str] = []
+        seen: set[str] = set()
+        for candidate in raw_candidates + [c.removeprefix("skillevaluator-") for c in raw_candidates]:
+            if candidate and candidate not in seen:
+                seen.add(candidate)
+                candidates.append(candidate)
+
         for candidate in candidates:
             if candidate in expected_case_ids:
                 return candidate
 
-    return _strip_arm_and_attempt_suffixes(name_part.removeprefix("skillevaluator-"))
+    return _strip_attempt_suffix(name_part)
 
 
 def _entry_id(reward: dict[str, Any], expected_case_ids: set[str] | None = None) -> str:
     if reward.get("entry_id"):
-        return _canonical_case_id(str(reward["entry_id"]), expected_case_ids)
+        raw_entry_id = str(reward["entry_id"]).strip()
+        if not reward.get("_result_entry_id"):
+            if expected_case_ids:
+                if raw_entry_id in expected_case_ids:
+                    return raw_entry_id
+                stripped_attempt = _strip_attempt_suffix(raw_entry_id)
+                if stripped_attempt in expected_case_ids:
+                    return stripped_attempt
+            else:
+                return _strip_attempt_suffix(raw_entry_id)
+        return _canonical_case_id(
+            raw_entry_id,
+            expected_case_ids,
+            strip_arm=not bool(reward.get("_arm_suffix")),
+        )
     trial_name = str(reward.get("_trial_name") or "")
     if trial_name:
-        return _canonical_case_id(trial_name.split("__", 1)[0], expected_case_ids)
+        trial_prefix = trial_name.split("__", 1)[0]
+        arm_suffix = str(reward.get("_arm_suffix") or "")
+        if arm_suffix:
+            attempt_stripped = _strip_attempt_suffix(trial_prefix)
+            if attempt_stripped.endswith(arm_suffix):
+                trial_prefix = attempt_stripped.removesuffix(arm_suffix)
+        return _canonical_case_id(
+            trial_prefix,
+            expected_case_ids,
+            strip_arm=not bool(arm_suffix),
+        )
     return "unknown"
 
 
@@ -2441,11 +2515,17 @@ def _annotate_security_attribution(
     without_rewards: list[dict[str, Any]],
     *,
     baseline_run: bool = True,
+    expected_case_ids: list[str] | set[str] | None = None,
 ) -> dict[str, Any]:
     """Annotate with-skill security findings with baseline-aware attribution."""
+    expected_id_set = (
+        {str(case_id).strip() for case_id in expected_case_ids if str(case_id).strip()}
+        if expected_case_ids is not None
+        else None
+    )
     baseline_by_case: dict[str, list[dict[str, Any]]] = {}
     for reward in without_rewards:
-        baseline_by_case.setdefault(_entry_id(reward), []).extend(_security_score_findings(reward))
+        baseline_by_case.setdefault(_entry_id(reward, expected_id_set), []).extend(_security_score_findings(reward))
 
     summary = {
         "likely_skill_related": 0,
@@ -2457,7 +2537,7 @@ def _annotate_security_attribution(
     }
 
     for reward in with_rewards:
-        entry_id = _entry_id(reward)
+        entry_id = _entry_id(reward, expected_id_set)
         details = reward.get("details")
         if not isinstance(details, dict):
             continue
@@ -2756,8 +2836,14 @@ def _save_trials(
     variant: str,
     agent_model: str | None = None,
     agent_model_source: str | None = None,
+    expected_case_ids: list[str] | set[str] | None = None,
 ) -> None:
     """Save per-trial reward.json and trajectory.json into the results directory."""
+    expected_id_set = (
+        {str(case_id).strip() for case_id in expected_case_ids if str(case_id).strip()}
+        if expected_case_ids is not None
+        else None
+    )
     trials_dir.mkdir(parents=True, exist_ok=True)
     for reward in rewards:
         trial_name, trial_root_name = _persisted_trial_name(reward)
@@ -2782,8 +2868,8 @@ def _save_trials(
         # Harbor trial once per step.  This also populates the canonical report's
         # existing trial_id field instead of inventing a second report schema.
         clean_reward["trial_id"] = trial_root_name
-        if not clean_reward.get("entry_id"):
-            clean_reward["entry_id"] = _entry_id(reward)
+        if not clean_reward.get("entry_id") or reward.get("_result_entry_id"):
+            clean_reward["entry_id"] = _entry_id(reward, expected_id_set)
         clean_reward["agent"] = agent
         if agent_model:
             clean_reward["model"] = agent_model
@@ -3095,6 +3181,7 @@ def collect_harbor_results(
         with_job_failure = ""
         with_execution: dict[str, Any] = {}
 
+        with_arm_suffix = "-with-skill" if not skip_baseline else ""
         if with_job_dir:
             with_job_ok, with_job_failure = validate_harbor_job_result(
                 with_job_dir / "result.json",
@@ -3103,7 +3190,9 @@ def collect_harbor_results(
             with_runtime_failures = _extract_agent_runtime_failures(with_job_dir)
             with_trial_failures = _extract_trial_failures(with_job_dir)
             preserve_partial = _can_preserve_partial_rewards(with_job_dir, with_trial_failures)
-            with_collected_rewards = _extract_rewards(with_job_dir) if with_job_ok or preserve_partial else []
+            with_collected_rewards = (
+                _extract_rewards(with_job_dir, arm_suffix=with_arm_suffix) if with_job_ok or preserve_partial else []
+            )
             with_rewards, invalid_score_failures = _partition_scoreable_rewards(with_collected_rewards)
             with_logical_rewards = _logical_attempt_rewards(with_rewards)
             with_trial_failures.extend(invalid_score_failures)
@@ -3147,6 +3236,7 @@ def collect_harbor_results(
                 variant="with_skill",
                 agent_model=agent_model,
                 agent_model_source=agent_model_source,
+                expected_case_ids=expected_case_ids,
             )
             (agent_dir / "with-skill" / "summary.json").write_text(
                 json.dumps(
@@ -3263,7 +3353,9 @@ def collect_harbor_results(
                 without_trial_failures = _extract_trial_failures(without_job_dir)
                 preserve_partial = _can_preserve_partial_rewards(without_job_dir, without_trial_failures)
                 without_collected_rewards = (
-                    _extract_rewards(without_job_dir) if without_job_ok or preserve_partial else []
+                    _extract_rewards(without_job_dir, arm_suffix="-without-skill")
+                    if without_job_ok or preserve_partial
+                    else []
                 )
                 without_rewards, invalid_score_failures = _partition_scoreable_rewards(without_collected_rewards)
                 without_logical_rewards = _logical_attempt_rewards(without_rewards)
@@ -3307,6 +3399,7 @@ def collect_harbor_results(
                     variant="without_skill",
                     agent_model=agent_model,
                     agent_model_source=agent_model_source,
+                    expected_case_ids=expected_case_ids,
                 )
                 (agent_dir / "without-skill" / "summary.json").write_text(
                     json.dumps(
@@ -3452,6 +3545,7 @@ def collect_harbor_results(
                 with_rewards,
                 without_rewards,
                 baseline_run=not skip_baseline,
+                expected_case_ids=expected_case_ids,
             )
             (agent_dir / "security_attribution.json").write_text(
                 json.dumps(security_attribution, indent=2), encoding="utf-8"
@@ -3466,6 +3560,7 @@ def collect_harbor_results(
                     variant="with_skill",
                     agent_model=agent_model,
                     agent_model_source=agent_model_source,
+                    expected_case_ids=expected_case_ids,
                 )
 
         agent_execution = _aggregate_execution([with_execution, without_execution])
diff --git a/tests/test_tier3_public_runtime.py b/tests/test_tier3_public_runtime.py
index 5f3ec033..09472bda 100644
--- a/tests/test_tier3_public_runtime.py
+++ b/tests/test_tier3_public_runtime.py
@@ -1375,45 +1375,41 @@ def test_strip_arm_and_attempt_suffixes(value: str, expected: str) -> None:
 @pytest.mark.parametrize(
     ("raw_input", "expected_ids", "expected_output"),
     [
-        # Bare case IDs without suffixes
+        # Bare case IDs without expected_case_ids preserve authored prefixes/suffixes
         ("case-001", None, "case-001"),
-        ("skillevaluator-case-001", None, "case-001"),
+        ("skillevaluator-case-001", None, "skillevaluator-case-001"),
         ("skillevaluator-case-001", {"case-001", "case-002"}, "case-001"),
-        # Dual-arm suffixes without expected_case_ids (fallback)
-        ("case-001-with-skill", None, "case-001"),
-        ("case-001-without-skill", None, "case-001"),
-        ("skillevaluator-case-001-with-skill", None, "case-001"),
-        ("skillevaluator-case-001-without-skill", None, "case-001"),
+        # Without expected_case_ids, authored -with-skill / -without-skill are preserved
+        ("case-001-with-skill", None, "case-001-with-skill"),
+        ("case-001-without-skill", None, "case-001-without-skill"),
+        ("skillevaluator-case-001-with-skill", None, "skillevaluator-case-001-with-skill"),
+        ("skillevaluator-case-001-without-skill", None, "skillevaluator-case-001-without-skill"),
         # Dual-arm suffixes with expected_case_ids matching
         ("case-001-with-skill", {"case-001", "case-002"}, "case-001"),
         ("case-001-without-skill", {"case-001", "case-002"}, "case-001"),
         ("skillevaluator-case-001-with-skill", {"case-001", "case-002"}, "case-001"),
         ("skillevaluator-case-001-without-skill", {"case-001", "case-002"}, "case-001"),
         # Namespaced task names (e.g., nvidia/...)
-        ("nvidia/skillevaluator-case-001", None, "case-001"),
+        ("nvidia/skillevaluator-case-001", None, "skillevaluator-case-001"),
         ("nvidia/skillevaluator-case-001", {"case-001"}, "case-001"),
-        ("nvidia/skillevaluator-case-001-with-skill", None, "case-001"),
+        ("nvidia/skillevaluator-case-001-with-skill", None, "skillevaluator-case-001-with-skill"),
         ("nvidia/skillevaluator-case-001-with-skill", {"case-001"}, "case-001"),
-        ("nvidia/skillevaluator-case-001-without-skill", None, "case-001"),
+        ("nvidia/skillevaluator-case-001-without-skill", None, "skillevaluator-case-001-without-skill"),
         ("nvidia/skillevaluator-case-001-without-skill", {"case-001"}, "case-001"),
         ("custom/repo/skillevaluator-case-002-with-skill", {"case-002"}, "case-002"),
         # Attempt suffixes combined with dual-arm suffixes in both orderings
-        ("case-001-with-skill-attempt1", None, "case-001"),
+        ("case-001-with-skill-attempt1", None, "case-001-with-skill"),
         ("case-001-with-skill-attempt1", {"case-001"}, "case-001"),
-        ("case-001-without-skill_attempt2", None, "case-001"),
+        ("case-001-without-skill_attempt2", None, "case-001-without-skill"),
         ("case-001-without-skill_attempt2", {"case-001"}, "case-001"),
-        ("case-001-attempt1-with-skill", None, "case-001"),
         ("case-001-attempt1-with-skill", {"case-001"}, "case-001"),
-        ("case-001_attempt2-without-skill", None, "case-001"),
         ("case-001_attempt2-without-skill", {"case-001"}, "case-001"),
-        ("skillevaluator-case-001-with-skill-attempt1", None, "case-001"),
+        ("skillevaluator-case-001-with-skill-attempt1", None, "skillevaluator-case-001-with-skill"),
         ("skillevaluator-case-001-with-skill-attempt1", {"case-001"}, "case-001"),
-        ("skillevaluator-case-001-attempt1-with-skill", None, "case-001"),
         ("skillevaluator-case-001-attempt1-with-skill", {"case-001"}, "case-001"),
         ("nvidia/skillevaluator-case-001-with-skill-attempt3", {"case-001"}, "case-001"),
         ("nvidia/skillevaluator-case-001-attempt3-with-skill", {"case-001"}, "case-001"),
-        ("nvidia/skillevaluator-case-001-without-skill_attempt4", None, "case-001"),
-        ("nvidia/skillevaluator-case-001_attempt4-without-skill", None, "case-001"),
+        ("nvidia/skillevaluator-case-001-without-skill_attempt4", None, "skillevaluator-case-001-without-skill"),
         # Legitimate case IDs ending with -with or -without preserved when in expected_case_ids
         ("case-with", {"case-with"}, "case-with"),
         ("case-without", {"case-without"}, "case-without"),
@@ -1444,6 +1440,11 @@ def test_strip_arm_and_attempt_suffixes(value: str, expected: str) -> None:
         ("", None, ""),
         ("   ", None, ""),
         ("", {"case-001"}, ""),
+        # Double arm suffixes must strip at most one arm suffix (never collapse authored -with-skill onto base case)
+        ("case-1-with-skill-with-skill", {"case-1"}, "case-1-with-skill-with-skill"),
+        ("case-1-with-skill-with-skill", {"case-1", "case-1-with-skill"}, "case-1-with-skill"),
+        ("case-1-with-skill-with-skill_attempt2", {"case-1", "case-1-with-skill"}, "case-1-with-skill"),
+        ("case-1-with-skill_attempt2-with-skill", {"case-1", "case-1-with-skill"}, "case-1-with-skill"),
     ],
 )
 def test_canonical_case_id_arm_stripping(
@@ -1496,6 +1497,239 @@ def test_stage_native_harbor_tasks_dual_arm_suffix(
     assert task["task"]["name"] == expected_task_name
 
 
+@pytest.mark.parametrize(
+    ("task_toml_body", "expected_task_name", "expected_metadata_name"),
+    [
+        (
+            'schema_version = "1.3"\n\n[task]\nname = """nvidia/case-001"""\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+        (
+            "schema_version = \"1.3\"\n\n[task]\nname = '''nvidia/case-001'''\n\n[environment]\n",
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+        (
+            'schema_version = "1.3"\n\n[metadata]\nname = "keep-metadata-name"\n\n'
+            '[task] # task identity\nauthors = [{ name = "Alice" }]\nname = "nvidia/case-001" # inline comment\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            "keep-metadata-name",
+        ),
+        (
+            'schema_version = "1.3"\n"task.name" = "root-quoted-dot-key"\n\n'
+            '[task.subtable]\nname = "subtable-name"\n\n'
+            '[task]\nkeywords = [\n  "alpha",\n  "beta",\n]\n'
+            'description = """\n[metadata]\n[task]\nname = "fake-inside-multiline"\n"""\n'
+            'name = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+        (
+            'schema_version = "1.3"\nmetadata = { name = "inline-metadata-name" }\n\n'
+            '[task]\nname = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            "inline-metadata-name",
+        ),
+        (
+            'schema_version = "1.3"\ntask.name = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+    ],
+)
+def test_stage_native_harbor_tasks_structural_toml_variants(
+    tmp_path: Path,
+    task_toml_body: str,
+    expected_task_name: str,
+    expected_metadata_name: str | None,
+) -> None:
+    """Verify structural [task].name update preserves valid TOML across multiline strings, comments, and [metadata].name."""
+    skill_dir = tmp_path / "target-skill"
+    skill_dir.mkdir()
+    (skill_dir / "SKILL.md").write_text("# Target Skill\n", encoding="utf-8")
+    task_dir = skill_dir / "evals" / "harbor" / "case-001"
+    task_dir.mkdir(parents=True)
+    (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
+    (task_dir / "task.toml").write_text(task_toml_body, encoding="utf-8")
+    tests_dir = task_dir / "tests"
+    tests_dir.mkdir()
+    (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+    staged = stage_native_harbor_tasks(
+        skill_dir,
+        tmp_path / "out",
+        grading_mode="custom_only",
+        arm_suffix="-with-skill",
+    )[0]
+    parsed = tomllib.loads((staged / "task.toml").read_text(encoding="utf-8"))
+    assert parsed["task"]["name"] == expected_task_name
+    if expected_metadata_name is not None:
+        assert parsed["metadata"]["name"] == expected_metadata_name
+    if "task.name" in parsed:
+        assert parsed["task.name"] == "root-quoted-dot-key"
+        assert parsed["task"]["subtable"]["name"] == "subtable-name"
+
+
+def test_append_native_task_name_suffix_sad_paths(tmp_path: Path) -> None:
+    """Verify _append_native_task_name_suffix safely no-ops on missing, corrupt, or non-string task names."""
+    from skillevaluator.tier3.harbor.adapter import _append_native_task_name_suffix
+
+    missing_dir = tmp_path / "missing"
+    missing_dir.mkdir()
+    _append_native_task_name_suffix(missing_dir, "-with-skill")
+    assert not (missing_dir / "task.toml").exists()
+
+    for idx, body in enumerate(
+        (
+            "[task\nname = 'unclosed'\n",
+            'schema_version = "1.3"\n[environment]\n',
+            'schema_version = "1.3"\n[task]\nname = 123\n',
+        )
+    ):
+        case_dir = tmp_path / f"sad-{idx}"
+        case_dir.mkdir()
+        (case_dir / "task.toml").write_text(body, encoding="utf-8")
+        _append_native_task_name_suffix(case_dir, "-with-skill")
+        assert (case_dir / "task.toml").read_text(encoding="utf-8") == body
+
+    preserved_dir = tmp_path / "preserved-entry"
+    preserved_dir.mkdir()
+    (preserved_dir / "task.toml").write_text(
+        '[metadata]\nentry_id = "authored-entry"\n\n[task]\nname = "nvidia/case-1"\n',
+        encoding="utf-8",
+    )
+    _append_native_task_name_suffix(preserved_dir, "")
+    assert tomllib.loads((preserved_dir / "task.toml").read_text(encoding="utf-8"))["task"]["name"] == "nvidia/case-1"
+    _append_native_task_name_suffix(preserved_dir, "-with-skill", entry_id="authored-entry")
+    parsed = tomllib.loads((preserved_dir / "task.toml").read_text(encoding="utf-8"))
+    assert parsed["task"]["name"] == "nvidia/case-1-with-skill"
+    assert parsed["metadata"]["entry_id"] == "authored-entry"
+
+
+def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom_only_collection(
+    tmp_path: Path,
+) -> None:
+    """Verify dual-arm staging and custom-only collection keep 'case-1' and 'case-1-with-skill' distinct."""
+    from skillevaluator.tier3.harbor.collector import collect_harbor_results
+
+    skill_dir = tmp_path / "target-skill"
+    skill_dir.mkdir()
+    (skill_dir / "SKILL.md").write_text("# Target Skill\n", encoding="utf-8")
+    harbor_dir = skill_dir / "evals" / "harbor"
+    for case_id in ("case-1", "case-1-with-skill"):
+        task_dir = harbor_dir / case_id
+        task_dir.mkdir(parents=True)
+        (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
+        (task_dir / "task.toml").write_text(
+            f'schema_version = "1.3"\n\n[task]\nname = "nvidia/{case_id}"\n\n[environment]\n',
+            encoding="utf-8",
+        )
+        tests_dir = task_dir / "tests"
+        tests_dir.mkdir()
+        (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+    with_staged = {
+        p.name: tomllib.loads((p / "task.toml").read_text(encoding="utf-8"))
+        for p in stage_native_harbor_tasks(
+            skill_dir,
+            tmp_path / "staged-with",
+            grading_mode="custom_only",
+            arm_suffix="-with-skill",
+        )
+    }
+    without_staged = {
+        p.name: tomllib.loads((p / "task.toml").read_text(encoding="utf-8"))
+        for p in stage_native_harbor_tasks(
+            skill_dir,
+            tmp_path / "staged-without",
+            grading_mode="custom_only",
+            with_skill=False,
+            arm_suffix="-without-skill",
+        )
+    }
+
+    assert with_staged["case-1"]["task"]["name"] == "nvidia/case-1-with-skill"
+    assert with_staged["case-1-with-skill"]["task"]["name"] == "nvidia/case-1-with-skill-with-skill"
+    assert without_staged["case-1"]["task"]["name"] == "nvidia/case-1-without-skill"
+    assert without_staged["case-1-with-skill"]["task"]["name"] == "nvidia/case-1-with-skill-without-skill"
+
+    jobs_dir = tmp_path / "jobs"
+    for variant, staged_map, scores in (
+        ("with", with_staged, {"case-1": 1.0, "case-1-with-skill": 0.9}),
+        ("without", without_staged, {"case-1": 0.2, "case-1-with-skill": 0.3}),
+    ):
+        job_dir = jobs_dir / f"target-skill-opencode-{variant}"
+        trial_names: list[str] = []
+        for idx, (case_id, parsed_toml) in enumerate(staged_map.items(), start=1):
+            trial_name = f"trial-{idx}__attempt1"
+            trial_names.append(trial_name)
+            trial_dir = job_dir / trial_name
+            trial_dir.mkdir(parents=True)
+            (trial_dir / "result.json").write_text(
+                json.dumps(
+                    {
+                        "trial_name": trial_name,
+                        "task_name": f"{parsed_toml['task']['name']}_attempt1",
+                        "verifier_result": {"rewards": {"reward": scores[case_id]}},
+                    }
+                ),
+                encoding="utf-8",
+            )
+        (job_dir / "result.json").write_text(
+            json.dumps(
+                {
+                    "n_total_trials": len(trial_names),
+                    "stats": {
+                        "n_trials": len(trial_names),
+                        "n_errors": 0,
+                        "evals": {
+                            "opencode": {
+                                "n_trials": len(trial_names),
+                                "n_errors": 0,
+                                "reward_stats": {"reward": {"1.0": trial_names}},
+                            }
+                        },
+                    },
+                }
+            ),
+            encoding="utf-8",
+        )
+
+    for idx, expected_ids in enumerate((["case-1", "case-1-with-skill"], None)):
+        result = collect_harbor_results(
+            skill_name="target-skill",
+            agents=["opencode"],
+            output_dir=tmp_path / f"results-{idx}",
+            jobs_dir=jobs_dir,
+            n_attempts=1,
+            expected_cases=2,
+            expected_case_ids=expected_ids,
+        )
+
+        assert result["execution_status"] == "succeeded"
+        agent_res = result["agents"]["opencode"]
+        assert set(agent_res["pass_at_k"]["with_skill"]["cases"]) == {"case-1", "case-1-with-skill"}
+        assert set(agent_res["pass_at_k"]["without_skill"]["cases"]) == {"case-1", "case-1-with-skill"}
+        assert agent_res["pass_at_k"]["with_skill"]["cases"]["case-1"]["best_score"] == 1.0
+        assert agent_res["pass_at_k"]["with_skill"]["cases"]["case-1-with-skill"]["best_score"] == 0.9
+        assert agent_res["pass_at_k"]["without_skill"]["cases"]["case-1"]["best_score"] == 0.2
+        assert agent_res["pass_at_k"]["without_skill"]["cases"]["case-1-with-skill"]["best_score"] == 0.3
+
+    # Sad path: if only 'case-1' is expected, 'case-1-with-skill' must be flagged as unexpected, not collapsed onto 'case-1'
+    partial_expected_result = collect_harbor_results(
+        skill_name="target-skill",
+        agents=["opencode"],
+        output_dir=tmp_path / "results-unexpected",
+        jobs_dir=jobs_dir,
+        n_attempts=1,
+        expected_cases=2,
+        expected_case_ids=["case-1"],
+    )
+    assert partial_expected_result["execution_status"] == "failed"
+    assert any("case-1-with-skill" in err for err in partial_expected_result["execution_errors"])
+
+
 def test_stage_native_harbor_tasks_type_safety(tmp_path: Path) -> None:
     """Verify that native staging functions reject non-string arm_suffix values."""
     skill_dir = tmp_path / "target-skill"
@@ -1540,4 +1774,3 @@ def test_generate_harbor_tasks_dual_arm_suffix(tmp_path: Path) -> None:
             evaluator_skill_path=skill_dir,
             arm_suffix=123,  # type: ignore[arg-type]
         )
-
diff --git a/tests/tier3/test_harbor_security_attribution.py b/tests/tier3/test_harbor_security_attribution.py
index da1ebd56..9647afc8 100644
--- a/tests/tier3/test_harbor_security_attribution.py
+++ b/tests/tier3/test_harbor_security_attribution.py
@@ -83,3 +83,25 @@ def test_security_attribution_avoids_skill_blame_without_baseline():
     finding = with_rewards[0]["details"]["security"]["findings"][0]
     assert finding["attribution"] == "unknown_no_baseline"
     assert summary["unknown_no_baseline"] == 1
+
+
+def test_security_attribution_preserves_distinct_authored_ids_without_expected_set():
+    """Verify authored IDs like case-1-with-skill and skillevaluator-case-1 do not collapse onto case-1."""
+    with_rewards = [
+        _reward("case-1", []),
+        _reward("case-1-with-skill", [_finding(target_skill_used_before=True)]),
+        _reward("skillevaluator-case-1", [_finding(target_skill_used_before=True)]),
+    ]
+    without_rewards = [
+        _reward("case-1", [_finding()]),
+        _reward("case-1-with-skill", []),
+        _reward("skillevaluator-case-1", []),
+    ]
+
+    summary = _annotate_security_attribution(with_rewards, without_rewards)
+
+    assert with_rewards[1]["details"]["security"]["findings"][0]["attribution"] == "likely_skill_related"
+    assert with_rewards[2]["details"]["security"]["findings"][0]["attribution"] == "likely_skill_related"
+    assert summary["likely_skill_related"] == 2
+    assert summary["likely_baseline_prompt_or_environment"] == 0
+    assert set(summary["cases"]) == {"case-1", "case-1-with-skill", "skillevaluator-case-1"}

From 5209b6f4fb5ba462ffb65479a24a61d6d178a0ca Mon Sep 17 00:00:00 2001
From: Karl Weinmeister <kweinmeister@google.com>
Date: Wed, 30 Sep 2026 17:33:06 -0400
Subject: [PATCH 4/4] fix(harbor): align native task.toml structural updates
 and [metadata].entry_id collection

Signed-off-by: Karl Weinmeister <kweinmeister@google.com>
---
 src/skillevaluator/tier3/harbor/adapter.py   | 293 +++++++++++-------
 src/skillevaluator/tier3/harbor/collector.py | 263 +++++++++++++---
 src/skillevaluator/tier3/harbor/runner.py    |  15 +-
 src/skillevaluator/tier3/toml_utils.py       |  15 +
 tests/test_tier3_public_runtime.py           | 305 +++++++++++++++++--
 5 files changed, 711 insertions(+), 180 deletions(-)

diff --git a/src/skillevaluator/tier3/harbor/adapter.py b/src/skillevaluator/tier3/harbor/adapter.py
index 974ef997..c080e76d 100644
--- a/src/skillevaluator/tier3/harbor/adapter.py
+++ b/src/skillevaluator/tier3/harbor/adapter.py
@@ -34,7 +34,12 @@
 from typing import Any
 from urllib.parse import unquote
 
-from skillevaluator.tier3.case_ids import safe_child, validate_case_ids, validate_output_directory_path
+from skillevaluator.tier3.case_ids import (
+    safe_child,
+    validate_case_id,
+    validate_case_ids,
+    validate_output_directory_path,
+)
 from skillevaluator.tier3.harbor import DEFAULT_LLM_VERIFIER_TIMEOUT_SEC
 from skillevaluator.tier3.harbor.secure_copy import (
     copy_file_secure,
@@ -49,7 +54,7 @@
     validate_provenance_key_outside,
     write_generated_output_marker,
 )
-from skillevaluator.tier3.toml_utils import toml_quote
+from skillevaluator.tier3.toml_utils import extract_toml_metadata_entry_id, toml_quote
 from skillevaluator.utils.process_environment import child_process_env
 from skillevaluator.utils.secure_fs import SecurePathError, SecureRoot
 
@@ -4216,16 +4221,15 @@ def _native_source_path_is_ignored(path: Path, native_dir: Path) -> bool:
 
 
 def _native_entry_id(task_dir: Path) -> str:
+    """Resolve and validate the canonical case ID for a native Harbor task directory."""
     task_toml = task_dir / "task.toml"
     try:
-        import tomllib
-
         data = tomllib.loads(task_toml.read_text(encoding="utf-8"))
     except Exception:
         return task_dir.name
-    metadata = data.get("metadata", {}) if isinstance(data, dict) else {}
-    if isinstance(metadata, dict) and metadata.get("entry_id"):
-        return str(metadata["entry_id"])
+    metadata = data.get("metadata") if isinstance(data, dict) else None
+    if isinstance(metadata, dict) and metadata.get("entry_id") is not None:
+        return validate_case_id(metadata["entry_id"])
     return task_dir.name
 
 
@@ -4240,6 +4244,34 @@ def _native_entry_id(task_dir: Path) -> str:
     _TOML_SINGLE_QUOTE,
 )
 _TOML_ESCAPE_PAIR_LEN = 2  # backslash + escaped character
+_TOML_DOTTED_KEY_PATTERN = (
+    r"""(?:"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*'|[A-Za-z0-9_-]+)"""
+    r"""(?:[ \t]*\.[ \t]*(?:"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*'|[A-Za-z0-9_-]+))*"""
+)
+_TOML_TABLE_HEADER_RE = re.compile(rf"\[(\[?)[ \t]*({_TOML_DOTTED_KEY_PATTERN})[ \t]*\]\]?[ \t]*(?:#[^\r\n]*)?\r?$")
+_TOML_KEY_ASSIGN_RE = re.compile(rf"({_TOML_DOTTED_KEY_PATTERN})[ \t]*=")
+
+
+@dataclass(frozen=True)
+class _TomlTableHeader:
+    """Represent a top-level [table] or [[array_of_tables]] header in a TOML document."""
+
+    is_array_table: bool
+    table_parts: tuple[str, ...]
+    line_start: int
+    line_end: int
+
+
+@dataclass(frozen=True)
+class _TomlKeyAssignment:
+    """Represent a key = value assignment at top level or inside an inline table."""
+
+    current_table: tuple[str, ...]
+    scope_stack: tuple[tuple[str, tuple[str, ...]], ...]
+    key_parts: tuple[str, ...]
+    full_path: tuple[str, ...]
+    key_start: int
+    val_start: int
 
 
 def _skip_toml_string_literal(content: str, start: int) -> int:
@@ -4270,22 +4302,26 @@ def _skip_toml_string_literal(content: str, start: int) -> int:
 
 
 def _parse_toml_dotted_key(raw: str) -> tuple[str, ...]:
-    """Parse a TOML key or table path into unquoted segment names."""
+    """Parse a TOML key or table path into unquoted, escape-decoded segment names."""
+    try:
+        parsed = tomllib.loads(f"{raw} = 1")
+    except Exception:
+        return ()
     parts: list[str] = []
-    for token in re.findall(r'"(?:\\.|[^"\\\r\n])*"|\'[^\'\r\n]*\'|[A-Za-z0-9_-]+', raw):
-        if (token.startswith('"') and token.endswith('"')) or (token.startswith("'") and token.endswith("'")):
-            parts.append(token[1:-1])
-        else:
-            parts.append(token)
+    curr: object = parsed
+    while isinstance(curr, dict) and len(curr) == 1:
+        key, curr = next(iter(curr.items()))
+        parts.append(key)
     return tuple(parts)
 
 
-def _find_toml_table_key_value_span(content: str, target_table: str, target_key: str) -> tuple[int, int] | None:
-    """Return the (start, end) character span of a string value for [target_table].target_key."""
+def _iter_toml_constructs(content: str) -> Iterable[_TomlTableHeader | _TomlKeyAssignment]:
+    """Yield top-level table headers and key assignments (including inside inline tables)."""
     n = len(content)
     i = 0
     current_table: tuple[str, ...] = ()
-    bracket_depth = 0
+    scope_stack: list[tuple[str, tuple[str, ...]]] = []
+    at_key_position = True
 
     while i < n:
         while i < n and content[i] in " \t":
@@ -4294,128 +4330,148 @@ def _find_toml_table_key_value_span(content: str, target_table: str, target_key:
             break
         if content[i] in "\r\n":
             i += 1
+            if not scope_stack:
+                at_key_position = True
             continue
         if content[i] == "#":
             while i < n and content[i] != "\n":
                 i += 1
             continue
 
-        if bracket_depth == 0 and content[i] == "[":
+        if not scope_stack and at_key_position and content[i] == "[":
             line_end = content.find("\n", i)
             if line_end == -1:
                 line_end = n
             line = content[i:line_end]
-            header_match = re.match(r"^\[(\[?)\s*([^\[\]#\r\n]+?)\s*\]\]?\s*(?:#.*)?\r?$", line)
+            header_match = _TOML_TABLE_HEADER_RE.match(line)
             if header_match:
                 is_array_table, raw_table = header_match.groups()
                 table_parts = _parse_toml_dotted_key(raw_table)
+                yield _TomlTableHeader(
+                    is_array_table=bool(is_array_table),
+                    table_parts=table_parts,
+                    line_start=i,
+                    line_end=line_end,
+                )
                 current_table = (f"[[{'.'.join(table_parts)}]]",) if is_array_table else table_parts
                 i = line_end + 1
+                at_key_position = True
                 continue
 
-        if bracket_depth == 0:
-            key_match = re.match(
-                r"""^(?:[A-Za-z0-9_-]+|"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*')(?:\s*\.\s*(?:[A-Za-z0-9_-]+|"(?:\\.|[^"\\\r\n])*"|'[^'\r\n]*'))*\s*=""",
-                content[i:],
-            )
+        if at_key_position and (not scope_stack or scope_stack[-1][0] == "table"):
+            key_match = _TOML_KEY_ASSIGN_RE.match(content, i)
             if key_match:
-                raw_lhs = content[i : i + key_match.end() - 1].strip()
+                raw_lhs = key_match.group(1)
                 key_parts = _parse_toml_dotted_key(raw_lhs)
-                full_path = (*current_table, *key_parts)
-                val_start = i + key_match.end()
+                base_path = scope_stack[-1][1] if scope_stack else current_table
+                full_path = (*base_path, *key_parts)
+                val_start = key_match.end()
                 while val_start < n and content[val_start] in " \t":
                     val_start += 1
-                if full_path == (target_table, target_key) and val_start < n and content[val_start] in "\"'":
-                    val_end = _skip_toml_string_literal(content, val_start)
-                    return val_start, val_end
+                yield _TomlKeyAssignment(
+                    current_table=current_table,
+                    scope_stack=tuple(scope_stack),
+                    key_parts=key_parts,
+                    full_path=full_path,
+                    key_start=i,
+                    val_start=val_start,
+                )
+                if val_start < n and content[val_start] == "{":
+                    scope_stack.append(("table", full_path))
+                    i = val_start + 1
+                    at_key_position = True
+                    continue
+                if val_start < n and content[val_start] == "[":
+                    scope_stack.append(("array", ()))
+                    i = val_start + 1
+                    at_key_position = False
+                    continue
                 i = val_start
-
-        while i < n:
-            ch = content[i]
-            if ch in "\"'":
-                i = _skip_toml_string_literal(content, i)
-                continue
-            if ch in "[({":
-                bracket_depth += 1
-                i += 1
-                continue
-            if ch in "])}":
-                bracket_depth = max(0, bracket_depth - 1)
-                i += 1
-                continue
-            if ch == "#":
-                while i < n and content[i] != "\n":
-                    i += 1
-                continue
-            if ch == "\n":
-                i += 1
-                if bracket_depth == 0:
-                    break
+                at_key_position = False
                 continue
+
+        ch = content[i]
+        if ch in "\"'":
+            i = _skip_toml_string_literal(content, i)
+            at_key_position = False
+        elif ch in "{[":
+            scope_stack.append(("array", ()))
+            i += 1
+            at_key_position = False
+        elif ch in "}]":
+            if scope_stack:
+                scope_stack.pop()
+            i += 1
+            at_key_position = False
+        elif ch == ",":
+            i += 1
+            at_key_position = bool(scope_stack and scope_stack[-1][0] == "table")
+        else:
             i += 1
+            at_key_position = False
 
+
+def _find_toml_table_key_value_span(content: str, target_table: str, target_key: str) -> tuple[int, int] | None:
+    """Return the (start, end) character span of a string value for [target_table].target_key."""
+    for construct in _iter_toml_constructs(content):
+        if (
+            isinstance(construct, _TomlKeyAssignment)
+            and construct.full_path == (target_table, target_key)
+            and construct.val_start < len(content)
+            and content[construct.val_start] in "\"'"
+        ):
+            val_end = _skip_toml_string_literal(content, construct.val_start)
+            return construct.val_start, val_end
     return None
 
 
 def _ensure_native_metadata_entry_id(content: str, entry_id: str) -> str:
     """Ensure [metadata].entry_id is present in a native task.toml document."""
+    quoted_id = _toml_quote(entry_id)
     n = len(content)
-    i = 0
-    bracket_depth = 0
-    while i < n:
-        while i < n and content[i] in " \t":
-            i += 1
-        if i >= n:
-            break
-        if content[i] in "\r\n":
-            i += 1
-            continue
-        if content[i] == "#":
-            while i < n and content[i] != "\n":
-                i += 1
-            continue
-        if bracket_depth == 0 and content[i] == "[":
-            line_end = content.find("\n", i)
-            if line_end == -1:
-                line_end = n
-            line = content[i:line_end]
-            header_match = re.match(r"^\[(\[?)\s*([^\[\]#\r\n]+?)\s*\]\]?\s*(?:#.*)?\r?$", line)
-            if header_match:
-                is_array_table, raw_table = header_match.groups()
-                if not is_array_table and _parse_toml_dotted_key(raw_table) == ("metadata",):
-                    insert_pos = line_end + 1 if line_end < n else n
-                    prefix = content[:insert_pos]
-                    if not prefix.endswith("\n"):
-                        prefix += "\n"
-                    return f"{prefix}entry_id = {_toml_quote(entry_id)}\n{content[insert_pos:]}"
-                i = line_end + 1
-                continue
-        while i < n:
-            ch = content[i]
-            if ch in "\"'":
-                i = _skip_toml_string_literal(content, i)
-                continue
-            if ch in "[({":
-                bracket_depth += 1
-                i += 1
-                continue
-            if ch in "])}":
-                bracket_depth = max(0, bracket_depth - 1)
-                i += 1
-                continue
-            if ch == "#":
-                while i < n and content[i] != "\n":
-                    i += 1
-                continue
-            if ch == "\n":
-                i += 1
-                if bracket_depth == 0:
-                    break
-                continue
-            i += 1
+    first_metadata_subtable_start: int | None = None
+
+    for construct in _iter_toml_constructs(content):
+        if isinstance(construct, _TomlTableHeader):
+            if not construct.is_array_table and construct.table_parts == ("metadata",):
+                insert_pos = construct.line_end + 1 if construct.line_end < n else n
+                prefix = content[:insert_pos]
+                if not prefix.endswith("\n"):
+                    prefix += "\n"
+                return f"{prefix}entry_id = {quoted_id}\n{content[insert_pos:]}"
+            if (
+                first_metadata_subtable_start is None
+                and len(construct.table_parts) >= 2
+                and construct.table_parts[0] == "metadata"
+            ):
+                first_metadata_subtable_start = construct.line_start
+        elif isinstance(construct, _TomlKeyAssignment):
+            if construct.full_path == ("metadata",) and construct.val_start < n and content[construct.val_start] == "{":
+                brace_pos = construct.val_start + 1
+                probe = brace_pos
+                while probe < n and content[probe] in " \t\r\n":
+                    probe += 1
+                if probe < n and content[probe] == "}":
+                    return f"{content[:brace_pos]} entry_id = {quoted_id} {content[brace_pos:]}"
+                return f"{content[:brace_pos]} entry_id = {quoted_id},{content[brace_pos:]}"
+            if (
+                not construct.scope_stack
+                and not construct.current_table
+                and len(construct.full_path) >= 2
+                and construct.full_path[0] == "metadata"
+            ):
+                return (
+                    f"{content[: construct.key_start]}metadata.entry_id = {quoted_id}\n{content[construct.key_start :]}"
+                )
+
+    if first_metadata_subtable_start is not None:
+        prefix = content[:first_metadata_subtable_start]
+        if prefix and not prefix.endswith("\n"):
+            prefix += "\n"
+        return f"{prefix}[metadata]\nentry_id = {quoted_id}\n\n{content[first_metadata_subtable_start:]}"
 
     suffix = "" if content.endswith("\n") else "\n"
-    return f"{content}{suffix}\n[metadata]\nentry_id = {_toml_quote(entry_id)}\n"
+    return f"{content}{suffix}\n[metadata]\nentry_id = {quoted_id}\n"
 
 
 def _append_native_task_name_suffix(
@@ -4446,23 +4502,23 @@ def _append_native_task_name_suffix(
     new_name = f"{old_name}{arm_suffix}"
     span = _find_toml_table_key_value_span(content, "task", "name")
     if span is None:
-        return
+        raise ValueError(f"Cannot update [task].name in native Harbor task config: {task_toml}")
     val_start, val_end = span
     new_content = f"{content[:val_start]}{_toml_quote(new_name)}{content[val_end:]}"
 
-    metadata = data.get("metadata")
-    effective_entry_id = entry_id or (
-        str(metadata["entry_id"]) if isinstance(metadata, dict) and metadata.get("entry_id") else task_dir.name
-    )
-    if effective_entry_id and not (isinstance(metadata, dict) and "entry_id" in metadata):
-        candidate_content = _ensure_native_metadata_entry_id(new_content, effective_entry_id)
-        try:
-            tomllib.loads(candidate_content)
-            new_content = candidate_content
-        except tomllib.TOMLDecodeError:
-            pass
+    existing_entry_id = extract_toml_metadata_entry_id(data)
+    effective_entry_id = entry_id or existing_entry_id or task_dir.name
+    if effective_entry_id and existing_entry_id is None:
+        new_content = _ensure_native_metadata_entry_id(new_content, effective_entry_id)
 
-    tomllib.loads(new_content)
+    try:
+        updated_data = tomllib.loads(new_content)
+    except tomllib.TOMLDecodeError as exc:
+        raise ValueError(f"Updated native Harbor task config is invalid TOML: {task_toml}") from exc
+    if updated_data.get("task", {}).get("name") != new_name:
+        raise ValueError(f"Failed to update [task].name in native Harbor task config: {task_toml}")
+    if effective_entry_id and extract_toml_metadata_entry_id(updated_data) != effective_entry_id:
+        raise ValueError(f"Failed to record [metadata].entry_id in native Harbor task config: {task_toml}")
     if new_content != content:
         task_toml.write_text(new_content, encoding="utf-8")
 
@@ -4851,6 +4907,7 @@ def _stage_native_harbor_tasks_into(
     validate_case_ids(path.name for path in source_task_dirs)
     for source_task_dir in source_task_dirs:
         _native_task_workdir(source_task_dir)
+    validate_case_ids(_native_entry_id(path) for path in source_task_dirs)
     _ = task_resources
     _ = agent_workdir
 
diff --git a/src/skillevaluator/tier3/harbor/collector.py b/src/skillevaluator/tier3/harbor/collector.py
index e807182d..67598e1a 100644
--- a/src/skillevaluator/tier3/harbor/collector.py
+++ b/src/skillevaluator/tier3/harbor/collector.py
@@ -17,6 +17,7 @@
 import shutil
 import stat
 import sys
+from collections.abc import Iterable, Mapping
 from fractions import Fraction
 from pathlib import Path
 from typing import Any
@@ -38,6 +39,7 @@
     score_definition,
 )
 from skillevaluator.tier3.output_provenance import write_output_file_atomically
+from skillevaluator.tier3.toml_utils import extract_toml_metadata_entry_id
 from skillevaluator.utils.redaction import is_sensitive_key, redact_sensitive_data, redact_sensitive_text
 from skillevaluator.utils.secure_fs import SecurePathError, SecureRoot, stat_is_link_or_reparse
 
@@ -47,6 +49,7 @@
 DEFAULT_DIAGNOSTIC_ARTIFACT_MAX_BYTES = 5 * 1024 * 1024
 DIAGNOSTIC_ARTIFACT_HARD_MAX_BYTES = 64 * 1024 * 1024
 REWARD_DIAGNOSTIC_STRING_MAX_CHARS = 8192
+MAX_STAGED_TASK_TOMLS = 512
 TRIAL_DIAGNOSTIC_ARTIFACTS = ("result.json", "config.json", "exception.txt", "trial.log")
 AGENT_LOG_ARTIFACTS = (
     "trajectory.json",
@@ -1577,12 +1580,130 @@ def _merge_constituent_default_reward_failure(
     )
 
 
-def _extract_rewards(job_dir: Path, *, arm_suffix: str = "") -> list[dict[str, Any]]:
+def _staged_task_entry_id_map(
+    output_dir: Path | None,
+    agent: str = "",
+    variant: str = "",
+    *,
+    arm_suffix: str = "",
+    max_tasks: int = MAX_STAGED_TASK_TOMLS,
+) -> dict[str, str]:
+    """Build an unambiguous mapping from staged task directory or task.name to [metadata].entry_id."""
+    if output_dir is None:
+        return {}
+    tasks_root = output_dir / "_harbor-tasks"
+    if not tasks_root.is_dir() or tasks_root.is_symlink():
+        return {}
+
+    search_roots: list[Path] = []
+    if agent and variant:
+        candidate = tasks_root / agent / variant
+        if candidate.is_dir() and not candidate.is_symlink():
+            search_roots.append(candidate)
+    if not search_roots:
+        search_roots.append(tasks_root)
+
+    mapping: dict[str, str] = {}
+    ambiguous: set[str] = set()
+
+    def _bind(key: str, entry_id: str) -> None:
+        if not key or key in ambiguous:
+            return
+        existing = mapping.get(key)
+        if existing is not None and existing != entry_id:
+            del mapping[key]
+            ambiguous.add(key)
+        else:
+            mapping[key] = entry_id
+
+    for root in search_roots:
+        try:
+            task_tomls = sorted(root.rglob("task.toml"))
+        except OSError:
+            continue
+        for task_toml in task_tomls[:max_tasks]:
+            if task_toml.is_symlink() or not task_toml.is_file():
+                continue
+            raw_toml = _read_bounded_text(task_toml)
+            if raw_toml is None:
+                continue
+            try:
+                import tomllib
+
+                parsed = tomllib.loads(raw_toml)
+            except Exception:
+                continue
+            if not isinstance(parsed, dict):
+                continue
+            raw_entry_id = extract_toml_metadata_entry_id(parsed)
+            entry_id = str(raw_entry_id).strip() if raw_entry_id is not None else ""
+            if not entry_id:
+                entry_id = task_toml.parent.name
+            _bind(task_toml.parent.name, entry_id)
+            task_table = parsed.get("task")
+            if isinstance(task_table, dict) and isinstance(task_table.get("name"), str):
+                task_name = task_table["name"].strip()
+                if task_name:
+                    _bind(task_name, entry_id)
+                    short_name = task_name.rsplit("/", 1)[-1]
+                    _bind(short_name, entry_id)
+                    if arm_suffix and short_name.endswith(arm_suffix):
+                        _bind(short_name.removesuffix(arm_suffix), entry_id)
+    return mapping
+
+
+def _extract_rewards(
+    job_dir: Path,
+    *,
+    arm_suffix: str = "",
+    task_entry_id_map: Mapping[str, str] | None = None,
+) -> list[dict[str, Any]]:
     """Extract reward.json from all trials in a job directory."""
     rewards: list[dict[str, Any]] = []
     scored_trial_roots: set[Path] = set()
     authoritative_trial_roots: set[Path] = set()
 
+    def _resolve_trial_name_entry_id(trial_name_val: str) -> str:
+        if not task_entry_id_map or not trial_name_val:
+            return ""
+        raw_prefix = trial_name_val.split("__", 1)[0]
+        candidates = [raw_prefix, _strip_attempt_suffix(raw_prefix)]
+        if arm_suffix:
+            for base in list(candidates):
+                if base.endswith(arm_suffix):
+                    stripped = base.removesuffix(arm_suffix)
+                    candidates.extend([stripped, _strip_attempt_suffix(stripped)])
+        for candidate in candidates:
+            if candidate in task_entry_id_map:
+                return task_entry_id_map[candidate]
+        return ""
+
+    def _populate_reward_entry_id(
+        reward_data: dict[str, Any],
+        result_payload: dict[str, Any] | None,
+        trial_name_val: str,
+    ) -> None:
+        if reward_data.get("entry_id"):
+            return
+        entry_id = ""
+        is_authoritative = False
+        if isinstance(result_payload, dict):
+            entry_id, is_authoritative = _resolve_harbor_result_entry_id(
+                result_payload,
+                arm_suffix=arm_suffix,
+                task_entry_id_map=task_entry_id_map,
+            )
+        if not entry_id:
+            entry_id = _resolve_trial_name_entry_id(trial_name_val)
+            if entry_id:
+                is_authoritative = True
+        if entry_id:
+            reward_data["entry_id"] = entry_id
+            if is_authoritative:
+                reward_data["_authoritative_entry_id"] = True
+            else:
+                reward_data["_result_entry_id"] = entry_id
+
     # Native multi-step Harbor tasks can persist an authoritative aggregate at
     # the trial root in addition to one reward file per step. Materialize that
     # single logical row first so both averages and pass@k use the same score.
@@ -1607,11 +1728,7 @@ def _extract_rewards(job_dir: Path, *, arm_suffix: str = "") -> list[dict[str, A
         data["_started_at"] = result.get("started_at")
         if arm_suffix:
             data["_arm_suffix"] = arm_suffix
-        if not data.get("entry_id"):
-            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
-            if entry_id:
-                data["entry_id"] = entry_id
-                data["_result_entry_id"] = entry_id
+        _populate_reward_entry_id(data, result, trial_name)
         traj_file = _reward_trajectory_path(trial_dir, None)
         if traj_file.exists():
             data["_has_trajectory"] = True
@@ -1645,16 +1762,14 @@ def _extract_rewards(job_dir: Path, *, arm_suffix: str = "") -> list[dict[str, A
                 if step_name:
                     data["_step_name"] = step_name
                 result_file = trial_dir / "result.json"
+                parsed_result: dict[str, Any] | None = None
                 if result_file.exists():
                     result = _read_json(result_file)
                     if isinstance(result, dict):
+                        parsed_result = result
                         _merge_constituent_default_reward_failure(data, result, trial_dir)
                         data["_started_at"] = result.get("started_at")
-                        if not data.get("entry_id"):
-                            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
-                            if entry_id:
-                                data["entry_id"] = entry_id
-                                data["_result_entry_id"] = entry_id
+                _populate_reward_entry_id(data, parsed_result, trial_name)
                 traj_file = _reward_trajectory_path(trial_dir, step_name)
                 if traj_file.exists():
                     data["_has_trajectory"] = True
@@ -1686,11 +1801,7 @@ def _extract_rewards(job_dir: Path, *, arm_suffix: str = "") -> list[dict[str, A
         data["_started_at"] = result.get("started_at")
         if arm_suffix:
             data["_arm_suffix"] = arm_suffix
-        if not data.get("entry_id"):
-            entry_id = _entry_id_from_harbor_result(result, arm_suffix=arm_suffix)
-            if entry_id:
-                data["entry_id"] = entry_id
-                data["_result_entry_id"] = entry_id
+        _populate_reward_entry_id(data, result, trial_name)
         traj_file = _reward_trajectory_path(trial_dir, None)
         if traj_file.exists():
             data["_has_trajectory"] = True
@@ -1787,21 +1898,29 @@ def _result_task_path_entry_id(task_path_value: Any) -> tuple[str, str]:
                 import tomllib
 
                 parsed = tomllib.loads(raw_toml)
-                metadata = parsed.get("metadata") if isinstance(parsed, dict) else None
-                if isinstance(metadata, dict) and metadata.get("entry_id"):
-                    return str(metadata["entry_id"]).strip(), task_dir.name
+                raw_entry_id = extract_toml_metadata_entry_id(parsed)
+                if raw_entry_id is not None:
+                    entry_id = str(raw_entry_id).strip()
+                    if entry_id:
+                        return entry_id, task_dir.name
             except Exception:
                 pass
     return "", task_dir.name
 
 
-def _entry_id_from_harbor_result(result: dict[str, Any], *, arm_suffix: str = "") -> str:
+def _resolve_harbor_result_entry_id(
+    result: dict[str, Any],
+    *,
+    arm_suffix: str = "",
+    task_entry_id_map: Mapping[str, str] | None = None,
+) -> tuple[str, bool]:
+    """Resolve (entry_id, is_authoritative) from a Harbor result.json payload."""
     fallback_dir_name = ""
     task_id = result.get("task_id")
     if isinstance(task_id, dict):
         meta_id, dir_name = _result_task_path_entry_id(task_id.get("path"))
         if meta_id:
-            return meta_id
+            return meta_id, True
         fallback_dir_name = dir_name or fallback_dir_name
 
     config = result.get("config")
@@ -1810,17 +1929,57 @@ def _entry_id_from_harbor_result(result: dict[str, Any], *, arm_suffix: str = ""
         if isinstance(task, dict):
             meta_id, dir_name = _result_task_path_entry_id(task.get("path"))
             if meta_id:
-                return meta_id
+                return meta_id, True
             fallback_dir_name = dir_name or fallback_dir_name
 
+    if fallback_dir_name and task_entry_id_map and fallback_dir_name in task_entry_id_map:
+        return task_entry_id_map[fallback_dir_name], True
+
     task_name = result.get("task_name")
     if isinstance(task_name, str) and task_name.strip():
-        short_name = _strip_attempt_suffix(task_name.strip().rsplit("/", 1)[-1])
-        if arm_suffix and short_name.endswith(arm_suffix):
-            short_name = short_name.removesuffix(arm_suffix)
-        return _strip_attempt_suffix(short_name)
+        raw_name = task_name.strip()
+        short_raw = raw_name.rsplit("/", 1)[-1]
+        attempt_stripped_raw = _strip_attempt_suffix(raw_name)
+        attempt_stripped_short = _strip_attempt_suffix(short_raw)
+
+        if task_entry_id_map:
+            lookup_keys = [raw_name, short_raw, attempt_stripped_raw, attempt_stripped_short]
+            if arm_suffix:
+                for base in (short_raw, attempt_stripped_short):
+                    if base.endswith(arm_suffix):
+                        without_arm = base.removesuffix(arm_suffix)
+                        lookup_keys.extend([without_arm, _strip_attempt_suffix(without_arm)])
+            for key in lookup_keys:
+                if key in task_entry_id_map:
+                    return task_entry_id_map[key], True
+
+        if arm_suffix:
+            if short_raw.endswith(arm_suffix):
+                return short_raw.removesuffix(arm_suffix), False
+            if attempt_stripped_short.endswith(arm_suffix):
+                return attempt_stripped_short.removesuffix(arm_suffix), False
+        return short_raw, False
 
-    return fallback_dir_name
+    if fallback_dir_name and task_entry_id_map and fallback_dir_name in task_entry_id_map:
+        return task_entry_id_map[fallback_dir_name], True
+    return fallback_dir_name, False
+
+
+def _entry_id_from_harbor_result(
+    result: dict[str, Any],
+    *,
+    arm_suffix: str = "",
+    task_entry_id_map: Mapping[str, str] | None = None,
+) -> str:
+    """Resolve the best-effort case ID from a Harbor trial result payload."""
+    entry_id, is_authoritative = _resolve_harbor_result_entry_id(
+        result,
+        arm_suffix=arm_suffix,
+        task_entry_id_map=task_entry_id_map,
+    )
+    if is_authoritative or not entry_id:
+        return entry_id
+    return _strip_attempt_suffix(entry_id)
 
 
 def _overall_score(reward: dict[str, Any]) -> float | None:
@@ -1937,9 +2096,21 @@ def _canonical_case_id(
     return _strip_attempt_suffix(name_part)
 
 
+def _normalize_expected_case_id_set(
+    expected_case_ids: Iterable[str] | None,
+) -> set[str] | None:
+    """Normalize optional expected case IDs into a stripped string set."""
+    if expected_case_ids is None:
+        return None
+    return {str(case_id).strip() for case_id in expected_case_ids if str(case_id).strip()}
+
+
 def _entry_id(reward: dict[str, Any], expected_case_ids: set[str] | None = None) -> str:
+    """Resolve the canonical case ID for a collected reward dictionary."""
     if reward.get("entry_id"):
         raw_entry_id = str(reward["entry_id"]).strip()
+        if reward.get("_authoritative_entry_id"):
+            return raw_entry_id
         if not reward.get("_result_entry_id"):
             if expected_case_ids:
                 if raw_entry_id in expected_case_ids:
@@ -2518,11 +2689,7 @@ def _annotate_security_attribution(
     expected_case_ids: list[str] | set[str] | None = None,
 ) -> dict[str, Any]:
     """Annotate with-skill security findings with baseline-aware attribution."""
-    expected_id_set = (
-        {str(case_id).strip() for case_id in expected_case_ids if str(case_id).strip()}
-        if expected_case_ids is not None
-        else None
-    )
+    expected_id_set = _normalize_expected_case_id_set(expected_case_ids)
     baseline_by_case: dict[str, list[dict[str, Any]]] = {}
     for reward in without_rewards:
         baseline_by_case.setdefault(_entry_id(reward, expected_id_set), []).extend(_security_score_findings(reward))
@@ -2839,11 +3006,7 @@ def _save_trials(
     expected_case_ids: list[str] | set[str] | None = None,
 ) -> None:
     """Save per-trial reward.json and trajectory.json into the results directory."""
-    expected_id_set = (
-        {str(case_id).strip() for case_id in expected_case_ids if str(case_id).strip()}
-        if expected_case_ids is not None
-        else None
-    )
+    expected_id_set = _normalize_expected_case_id_set(expected_case_ids)
     trials_dir.mkdir(parents=True, exist_ok=True)
     for reward in rewards:
         trial_name, trial_root_name = _persisted_trial_name(reward)
@@ -3190,8 +3353,20 @@ def collect_harbor_results(
             with_runtime_failures = _extract_agent_runtime_failures(with_job_dir)
             with_trial_failures = _extract_trial_failures(with_job_dir)
             preserve_partial = _can_preserve_partial_rewards(with_job_dir, with_trial_failures)
+            with_task_entry_ids = _staged_task_entry_id_map(
+                output_dir,
+                agent,
+                "with",
+                arm_suffix=with_arm_suffix,
+            )
             with_collected_rewards = (
-                _extract_rewards(with_job_dir, arm_suffix=with_arm_suffix) if with_job_ok or preserve_partial else []
+                _extract_rewards(
+                    with_job_dir,
+                    arm_suffix=with_arm_suffix,
+                    task_entry_id_map=with_task_entry_ids,
+                )
+                if with_job_ok or preserve_partial
+                else []
             )
             with_rewards, invalid_score_failures = _partition_scoreable_rewards(with_collected_rewards)
             with_logical_rewards = _logical_attempt_rewards(with_rewards)
@@ -3352,8 +3527,18 @@ def collect_harbor_results(
                 without_runtime_failures = _extract_agent_runtime_failures(without_job_dir)
                 without_trial_failures = _extract_trial_failures(without_job_dir)
                 preserve_partial = _can_preserve_partial_rewards(without_job_dir, without_trial_failures)
+                without_task_entry_ids = _staged_task_entry_id_map(
+                    output_dir,
+                    agent,
+                    "without",
+                    arm_suffix="-without-skill",
+                )
                 without_collected_rewards = (
-                    _extract_rewards(without_job_dir, arm_suffix="-without-skill")
+                    _extract_rewards(
+                        without_job_dir,
+                        arm_suffix="-without-skill",
+                        task_entry_id_map=without_task_entry_ids,
+                    )
                     if without_job_ok or preserve_partial
                     else []
                 )
diff --git a/src/skillevaluator/tier3/harbor/runner.py b/src/skillevaluator/tier3/harbor/runner.py
index 35615a17..02e24dec 100644
--- a/src/skillevaluator/tier3/harbor/runner.py
+++ b/src/skillevaluator/tier3/harbor/runner.py
@@ -43,6 +43,7 @@
 from skillevaluator.tier3.evals_config import EvalsConfigError, load_evals_config
 from skillevaluator.tier3.harbor.adapter import (
     _VERIFIER_JUDGE_MODEL_ENV_VARS,
+    _native_entry_id,
     _prevalidate_baseline_skill_candidates,
     build_eval_base_image,
     find_evals_file,
@@ -2328,6 +2329,7 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
             )
     agent_task_dirs: dict[str, tuple[Path, Path | None]] = {}
     expected_task_names: list[str] | None = None
+    expected_case_ids: list[str] | None = None
     reporter.emit(
         ProgressEvent(
             stage="with-skill-tasks",
@@ -2365,9 +2367,11 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
                 arm_suffix=with_arm_suffix,
             )
             task_names = [task.name for task in task_paths]
+            case_ids = [_native_entry_id(task) for task in task_paths]
             if expected_task_names is None:
                 expected_task_names = task_names
-            elif task_names != expected_task_names:
+                expected_case_ids = case_ids
+            elif task_names != expected_task_names or case_ids != expected_case_ids:
                 raise ValueError(f"Generated task cases differ for agent {agent}")
             agent_task_dirs[agent] = (with_dir, without_dir)
         reporter.emit(ProgressEvent(stage="with-skill-tasks", state="ready", detail="task inputs staged"))
@@ -2415,6 +2419,7 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
         return _persist_pre_execution_failure([str(exc)])
 
     task_names = expected_task_names or []
+    case_ids = expected_case_ids or task_names
     expected_trials = len(task_names) * n_attempts
     variants = 1 if skip_baseline else 2
     matrix_trials = expected_trials * len(agents) * variants
@@ -2431,7 +2436,7 @@ def _persist_pre_execution_failure(errors: list[str]) -> dict[str, Any]:
             agent_models=tuple((agent, model_resolution[agent]["model"]) for agent in agents),
             provider=provider.provider,
             task_count=len(task_names),
-            case_count=len(task_names),
+            case_count=len(case_ids),
             attempts=n_attempts,
             baseline=not skip_baseline,
             concurrency=n_concurrent,
@@ -2590,8 +2595,8 @@ def _emit_started_agents() -> None:
             n_attempts=n_attempts,
             pass_threshold=float(pass_threshold),
             stop_on_pass=bool(stop_on_pass),
-            expected_cases=len(task_names),
-            expected_case_ids=task_names,
+            expected_cases=len(case_ids),
+            expected_case_ids=case_ids,
             # Early-stopped cases legitimately use fewer trials than the
             # n_attempts maximum; per-case coverage is validated instead.
             expected_trials=None if stop_on_pass else expected_trials,
@@ -2604,7 +2609,7 @@ def _emit_started_agents() -> None:
         _emit_run_finished("failed", "result collection failed")
         raise
     reporter.emit(ProgressEvent(stage="collection", state="complete", detail="Harbor results collected"))
-    dataset_truth = _persist_dataset_truth(run_dir, fallback_task_ids=task_names)
+    dataset_truth = _persist_dataset_truth(run_dir, fallback_task_ids=case_ids)
     results.update(
         {
             "skill_name": skill_path.name,
diff --git a/src/skillevaluator/tier3/toml_utils.py b/src/skillevaluator/tier3/toml_utils.py
index f479d4d8..16da7779 100644
--- a/src/skillevaluator/tier3/toml_utils.py
+++ b/src/skillevaluator/tier3/toml_utils.py
@@ -6,6 +6,7 @@
 from __future__ import annotations
 
 import json
+from collections.abc import Mapping
 
 
 def toml_quote(value: str) -> str:
@@ -15,3 +16,17 @@ def toml_quote(value: str) -> str:
     if any(0xD800 <= ord(character) <= 0xDFFF for character in value):
         raise ValueError("TOML strings must not contain surrogate code points")
     return json.dumps(value, ensure_ascii=False).replace("\x7f", "\\u007F")
+
+
+def extract_toml_metadata_entry_id(parsed: object) -> str | None:
+    """Extract a non-empty, non-boolean [metadata].entry_id string from parsed TOML data."""
+    if not isinstance(parsed, Mapping):
+        return None
+    metadata = parsed.get("metadata")
+    if not isinstance(metadata, Mapping) or "entry_id" not in metadata:
+        return None
+    raw_entry_id = metadata["entry_id"]
+    if raw_entry_id is None or isinstance(raw_entry_id, bool):
+        return None
+    stripped = str(raw_entry_id).strip()
+    return stripped or None
diff --git a/tests/test_tier3_public_runtime.py b/tests/test_tier3_public_runtime.py
index 09472bda..ce53241a 100644
--- a/tests/test_tier3_public_runtime.py
+++ b/tests/test_tier3_public_runtime.py
@@ -1531,11 +1531,44 @@ def test_stage_native_harbor_tasks_dual_arm_suffix(
             "nvidia/case-001-with-skill",
             "inline-metadata-name",
         ),
+        (
+            'schema_version = "1.3"\nmetadata = {}\n\n'
+            'task = { authors = [{ name = "Alice" }], name = "nvidia/case-001", tags = ["a", "b"] }\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+        (
+            'schema_version = "1.3"\nmetadata.name = "dotted-metadata-name"\n'
+            'task = { subtable = { name = "subtable-name" }, name = "nvidia/case-001" }\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            "dotted-metadata-name",
+        ),
+        (
+            'schema_version = "1.3"\n["\\u006detadata"]\nname = "escaped-metadata-name"\n\n'
+            '["\\u0074ask"]\n"\\u006eame" = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            "escaped-metadata-name",
+        ),
+        (
+            'schema_version = "1.3"\n"\\u0074ask"."\\u006eame" = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
         (
             'schema_version = "1.3"\ntask.name = "nvidia/case-001"\n\n[environment]\n',
             "nvidia/case-001-with-skill",
             None,
         ),
+        (
+            'schema_version = "1.3"\n\n[metadata.extra]\nfoo = "bar"\n\n[task]\nname = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
+        (
+            'schema_version = "1.3"\n\n[[metadata.items]]\nname = "item1"\n\n[task]\nname = "nvidia/case-001"\n\n[environment]\n',
+            "nvidia/case-001-with-skill",
+            None,
+        ),
     ],
 )
 def test_stage_native_harbor_tasks_structural_toml_variants(
@@ -1544,7 +1577,7 @@ def test_stage_native_harbor_tasks_structural_toml_variants(
     expected_task_name: str,
     expected_metadata_name: str | None,
 ) -> None:
-    """Verify structural [task].name update preserves valid TOML across multiline strings, comments, and [metadata].name."""
+    """Verify structural [task].name update preserves valid TOML across inline tables, escaped keys, and [metadata]."""
     skill_dir = tmp_path / "target-skill"
     skill_dir.mkdir()
     (skill_dir / "SKILL.md").write_text("# Target Skill\n", encoding="utf-8")
@@ -1564,6 +1597,7 @@ def test_stage_native_harbor_tasks_structural_toml_variants(
     )[0]
     parsed = tomllib.loads((staged / "task.toml").read_text(encoding="utf-8"))
     assert parsed["task"]["name"] == expected_task_name
+    assert parsed["metadata"]["entry_id"] == "case-001"
     if expected_metadata_name is not None:
         assert parsed["metadata"]["name"] == expected_metadata_name
     if "task.name" in parsed:
@@ -1572,7 +1606,7 @@ def test_stage_native_harbor_tasks_structural_toml_variants(
 
 
 def test_append_native_task_name_suffix_sad_paths(tmp_path: Path) -> None:
-    """Verify _append_native_task_name_suffix safely no-ops on missing, corrupt, or non-string task names."""
+    """Verify _append_native_task_name_suffix safely handles missing/corrupt/non-string task names and fails closed when [metadata].entry_id cannot be written."""
     from skillevaluator.tier3.harbor.adapter import _append_native_task_name_suffix
 
     missing_dir = tmp_path / "missing"
@@ -1606,34 +1640,59 @@ def test_append_native_task_name_suffix_sad_paths(tmp_path: Path) -> None:
     assert parsed["task"]["name"] == "nvidia/case-1-with-skill"
     assert parsed["metadata"]["entry_id"] == "authored-entry"
 
+    uninjectable_dir = tmp_path / "uninjectable-metadata"
+    uninjectable_dir.mkdir()
+    (uninjectable_dir / "task.toml").write_text(
+        'metadata = "not-a-table"\n\n[task]\nname = "nvidia/case-1"\n',
+        encoding="utf-8",
+    )
+    with pytest.raises(ValueError, match="Updated native Harbor task config is invalid TOML"):
+        _append_native_task_name_suffix(uninjectable_dir, "-with-skill")
+
+    mismatched_dir = tmp_path / "mismatched-metadata"
+    mismatched_dir.mkdir()
+    (mismatched_dir / "task.toml").write_text(
+        '[metadata]\nentry_id = "existing-id"\n\n[task]\nname = "nvidia/case-1"\n',
+        encoding="utf-8",
+    )
+    with pytest.raises(ValueError, match=r"Failed to record \[metadata\]\.entry_id"):
+        _append_native_task_name_suffix(mismatched_dir, "-with-skill", entry_id="conflicting-id")
+
 
 def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom_only_collection(
     tmp_path: Path,
 ) -> None:
-    """Verify dual-arm staging and custom-only collection keep 'case-1' and 'case-1-with-skill' distinct."""
+    """Verify dual-arm staging and custom-only collection keep 'case-1', 'case-1-with-skill', and 'retry-attempt1' distinct across inline and standard task.toml syntax."""
     from skillevaluator.tier3.harbor.collector import collect_harbor_results
 
     skill_dir = tmp_path / "target-skill"
     skill_dir.mkdir()
     (skill_dir / "SKILL.md").write_text("# Target Skill\n", encoding="utf-8")
     harbor_dir = skill_dir / "evals" / "harbor"
-    for case_id in ("case-1", "case-1-with-skill"):
+    task_toml_templates = {
+        "case-1": 'schema_version = "1.3"\n\n["\\u0074ask"]\n"\\u006eame" = "nvidia/case-1"\n\n[environment]\n',
+        "case-1-with-skill": 'schema_version = "1.3"\ntask = { name = "nvidia/case-1-with-skill" }\n\n[environment]\n',
+        "retry-attempt1": 'schema_version = "1.3"\n[task]\nname = "nvidia/skillevaluator-retry-attempt1"\n\n[environment]\n',
+    }
+    for case_id, toml_body in task_toml_templates.items():
         task_dir = harbor_dir / case_id
         task_dir.mkdir(parents=True)
         (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
-        (task_dir / "task.toml").write_text(
-            f'schema_version = "1.3"\n\n[task]\nname = "nvidia/{case_id}"\n\n[environment]\n',
-            encoding="utf-8",
-        )
+        (task_dir / "task.toml").write_text(toml_body, encoding="utf-8")
         tests_dir = task_dir / "tests"
         tests_dir.mkdir()
         (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
 
+    # Stage into results-1/_harbor-tasks/opencode/{with,without} so idx=1 (expected_case_ids=None)
+    # tests authoritative [metadata].entry_id lookup via _staged_task_entry_id_map, while idx=0
+    # (expected_case_ids=[...], no _harbor-tasks in results-0) tests deferred attempt-suffix stripping
+    # on raw result["task_name"] protected by expected_case_ids.
+    staged_root = tmp_path / "results-1" / "_harbor-tasks" / "opencode"
     with_staged = {
         p.name: tomllib.loads((p / "task.toml").read_text(encoding="utf-8"))
         for p in stage_native_harbor_tasks(
             skill_dir,
-            tmp_path / "staged-with",
+            staged_root / "with",
             grading_mode="custom_only",
             arm_suffix="-with-skill",
         )
@@ -1642,7 +1701,7 @@ def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom
         p.name: tomllib.loads((p / "task.toml").read_text(encoding="utf-8"))
         for p in stage_native_harbor_tasks(
             skill_dir,
-            tmp_path / "staged-without",
+            staged_root / "without",
             grading_mode="custom_only",
             with_skill=False,
             arm_suffix="-without-skill",
@@ -1651,13 +1710,15 @@ def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom
 
     assert with_staged["case-1"]["task"]["name"] == "nvidia/case-1-with-skill"
     assert with_staged["case-1-with-skill"]["task"]["name"] == "nvidia/case-1-with-skill-with-skill"
+    assert with_staged["retry-attempt1"]["task"]["name"] == "nvidia/skillevaluator-retry-attempt1-with-skill"
     assert without_staged["case-1"]["task"]["name"] == "nvidia/case-1-without-skill"
     assert without_staged["case-1-with-skill"]["task"]["name"] == "nvidia/case-1-with-skill-without-skill"
+    assert without_staged["retry-attempt1"]["task"]["name"] == "nvidia/skillevaluator-retry-attempt1-without-skill"
 
     jobs_dir = tmp_path / "jobs"
     for variant, staged_map, scores in (
-        ("with", with_staged, {"case-1": 1.0, "case-1-with-skill": 0.9}),
-        ("without", without_staged, {"case-1": 0.2, "case-1-with-skill": 0.3}),
+        ("with", with_staged, {"case-1": 1.0, "case-1-with-skill": 0.9, "retry-attempt1": 0.8}),
+        ("without", without_staged, {"case-1": 0.2, "case-1-with-skill": 0.3, "retry-attempt1": 0.4}),
     ):
         job_dir = jobs_dir / f"target-skill-opencode-{variant}"
         trial_names: list[str] = []
@@ -1666,11 +1727,17 @@ def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom
             trial_names.append(trial_name)
             trial_dir = job_dir / trial_name
             trial_dir.mkdir(parents=True)
+            # Test both raw task_name (no _attempt1 suffix) and _attempt1 suffixed task_name
+            task_name_val = (
+                parsed_toml["task"]["name"]
+                if case_id == "retry-attempt1"
+                else f"{parsed_toml['task']['name']}_attempt1"
+            )
             (trial_dir / "result.json").write_text(
                 json.dumps(
                     {
                         "trial_name": trial_name,
-                        "task_name": f"{parsed_toml['task']['name']}_attempt1",
+                        "task_name": task_name_val,
                         "verifier_result": {"rewards": {"reward": scores[case_id]}},
                     }
                 ),
@@ -1696,25 +1763,31 @@ def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom
             encoding="utf-8",
         )
 
-    for idx, expected_ids in enumerate((["case-1", "case-1-with-skill"], None)):
+    for idx, expected_ids in enumerate((["case-1", "case-1-with-skill", "retry-attempt1"], None)):
         result = collect_harbor_results(
             skill_name="target-skill",
             agents=["opencode"],
             output_dir=tmp_path / f"results-{idx}",
             jobs_dir=jobs_dir,
             n_attempts=1,
-            expected_cases=2,
+            expected_cases=3,
             expected_case_ids=expected_ids,
         )
 
         assert result["execution_status"] == "succeeded"
         agent_res = result["agents"]["opencode"]
-        assert set(agent_res["pass_at_k"]["with_skill"]["cases"]) == {"case-1", "case-1-with-skill"}
-        assert set(agent_res["pass_at_k"]["without_skill"]["cases"]) == {"case-1", "case-1-with-skill"}
+        assert set(agent_res["pass_at_k"]["with_skill"]["cases"]) == {"case-1", "case-1-with-skill", "retry-attempt1"}
+        assert set(agent_res["pass_at_k"]["without_skill"]["cases"]) == {
+            "case-1",
+            "case-1-with-skill",
+            "retry-attempt1",
+        }
         assert agent_res["pass_at_k"]["with_skill"]["cases"]["case-1"]["best_score"] == 1.0
         assert agent_res["pass_at_k"]["with_skill"]["cases"]["case-1-with-skill"]["best_score"] == 0.9
+        assert agent_res["pass_at_k"]["with_skill"]["cases"]["retry-attempt1"]["best_score"] == 0.8
         assert agent_res["pass_at_k"]["without_skill"]["cases"]["case-1"]["best_score"] == 0.2
         assert agent_res["pass_at_k"]["without_skill"]["cases"]["case-1-with-skill"]["best_score"] == 0.3
+        assert agent_res["pass_at_k"]["without_skill"]["cases"]["retry-attempt1"]["best_score"] == 0.4
 
     # Sad path: if only 'case-1' is expected, 'case-1-with-skill' must be flagged as unexpected, not collapsed onto 'case-1'
     partial_expected_result = collect_harbor_results(
@@ -1723,13 +1796,209 @@ def test_dual_arm_native_tasks_with_authored_arm_suffix_do_not_collide_in_custom
         output_dir=tmp_path / "results-unexpected",
         jobs_dir=jobs_dir,
         n_attempts=1,
-        expected_cases=2,
+        expected_cases=3,
         expected_case_ids=["case-1"],
     )
     assert partial_expected_result["execution_status"] == "failed"
     assert any("case-1-with-skill" in err for err in partial_expected_result["execution_errors"])
 
 
+def test_stage_native_harbor_tasks_validates_metadata_entry_ids_sad_paths(tmp_path: Path) -> None:
+    """Verify native staging rejects unsafe or duplicate [metadata].entry_id values before mutating output_dir."""
+    # 1. Unsafe metadata.entry_id (e.g. path traversal or boolean)
+    for idx, bad_entry_id in enumerate(('"../escape"', "true", '""')):
+        skill_dir = tmp_path / f"bad-skill-{idx}"
+        skill_dir.mkdir()
+        (skill_dir / "SKILL.md").write_text("# Bad Skill\n", encoding="utf-8")
+        task_dir = skill_dir / "evals" / "harbor" / "physical-case"
+        task_dir.mkdir(parents=True)
+        (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
+        (task_dir / "task.toml").write_text(
+            f'[metadata]\nentry_id = {bad_entry_id}\n\n[task]\nname = "nvidia/physical-case"\n',
+            encoding="utf-8",
+        )
+        tests_dir = task_dir / "tests"
+        tests_dir.mkdir()
+        (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+        out_dir = tmp_path / f"out-bad-{idx}"
+        with pytest.raises(ValueError, match="case id"):
+            stage_native_harbor_tasks(skill_dir, out_dir, grading_mode="custom_only")
+        assert not out_dir.exists()
+
+    # 2. Duplicate metadata.entry_id across two distinct physical task directories
+    dup_skill = tmp_path / "dup-skill"
+    dup_skill.mkdir()
+    (dup_skill / "SKILL.md").write_text("# Dup Skill\n", encoding="utf-8")
+    for folder in ("physical-a", "physical-b"):
+        task_dir = dup_skill / "evals" / "harbor" / folder
+        task_dir.mkdir(parents=True)
+        (task_dir / "instruction.md").write_text("Instruction\n", encoding="utf-8")
+        (task_dir / "task.toml").write_text(
+            '[metadata]\nentry_id = "shared-authored-id"\n\n[task]\nname = "nvidia/task"\n',
+            encoding="utf-8",
+        )
+        tests_dir = task_dir / "tests"
+        tests_dir.mkdir()
+        (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+    dup_out = tmp_path / "out-dup"
+    with pytest.raises(ValueError, match=r"duplicate or cross-platform colliding case id.*shared-authored-id"):
+        stage_native_harbor_tasks(dup_skill, dup_out, grading_mode="custom_only")
+    assert not dup_out.exists()
+
+
+def test_native_harbor_metadata_entry_id_seam_across_runner_collector_and_dataset_snapshot(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Verify native custom-only tasks with [metadata].entry_id != directory name align across runner, collector, and dataset_snapshot.json."""
+    from skillevaluator.provider_config import ProviderConfig
+    from skillevaluator.tier3.harbor import runner, runtime_preflight
+
+    skill_dir = tmp_path / "target-skill"
+    skill_dir.mkdir()
+    (skill_dir / "SKILL.md").write_text("---\nname: target-skill\ndescription: demo\n---\n# Body\n", encoding="utf-8")
+    (skill_dir / "evals").mkdir()
+    (skill_dir / "evals" / "config.yaml").write_text(
+        "schema_version: 1\nharbor:\n  task_source: native_harbor\ngrading:\n  mode: custom_only\n",
+        encoding="utf-8",
+    )
+    for folder, authored_entry_id in (("physical-case", '"authored-entry"'), ("zero-case", "0")):
+        task_dir = skill_dir / "evals" / "harbor" / folder
+        task_dir.mkdir(parents=True)
+        (task_dir / "instruction.md").write_text("Solve the task.\n", encoding="utf-8")
+        (task_dir / "task.toml").write_text(
+            f'schema_version = "1.3"\n\n[metadata]\nentry_id = {authored_entry_id}\n\n'
+            f'[task]\nname = "nvidia/{folder}"\n\n[environment]\n',
+            encoding="utf-8",
+        )
+        tests_dir = task_dir / "tests"
+        tests_dir.mkdir()
+        (tests_dir / "test.sh").write_text("#!/bin/bash\nexit 0\n", encoding="utf-8")
+
+    output_dir = tmp_path / "eval-out"
+    provider = ProviderConfig(
+        provider="nv_build",
+        model="nvidia/nemotron-3-nano-30b-a3b",
+        api_key="nvapi-test",
+        base_url="https://integrate.api.nvidia.com/v1",
+        litellm_model="nvidia_nim/nvidia/nemotron-3-nano-30b-a3b",
+    )
+    monkeypatch.setattr(runner, "resolve_llm_provider", lambda: provider)
+    monkeypatch.setattr(runner, "_check_prerequisites", lambda **_kwargs: [])
+    monkeypatch.setattr(
+        runtime_preflight,
+        "probe_model",
+        lambda selected_provider: runtime_preflight.ModelProbeResult(
+            True,
+            selected_provider.provider,
+            selected_provider.model,
+            "ok",
+        ),
+    )
+
+    seen_include_task_names: list[list[str] | None] = []
+
+    def fake_run_harbor(
+        *,
+        dataset: Path,
+        agent: str,
+        job_name: str,
+        env_mode: str,
+        model: str,
+        jobs_dir: Path,
+        run_env: dict[str, str],
+        n_attempts: int,
+        n_concurrent: int,
+        timeout_multiplier: float,
+        override_cpus: int | None,
+        override_memory_mb: int | None,
+        override_storage_mb: int | None,
+        expected_trials: int,
+        agent_import_path: str | None = None,
+        verifier_env: object = None,
+        include_task_names: list[str] | None = None,
+    ) -> tuple[bool, str]:
+        seen_include_task_names.append(include_task_names)
+        job_dir = jobs_dir / job_name
+        is_with = "-with-" in job_name or job_name.endswith("-with")
+        arm_suffix = "-with-skill" if is_with else "-without-skill"
+        selected_folders = include_task_names or ["physical-case", "zero-case"]
+        trial_names: list[str] = []
+        for idx, folder in enumerate(selected_folders, start=1):
+            trial_name = f"trial-{idx}-{folder}__attempt1"
+            trial_names.append(trial_name)
+            trial_dir = job_dir / trial_name
+            trial_dir.mkdir(parents=True, exist_ok=True)
+            # Notice result.json only has task_name derived from the staged [task].name (physical-case),
+            # not [metadata].entry_id, exercising collector's staged task.toml entry_id lookup!
+            (trial_dir / "result.json").write_text(
+                json.dumps(
+                    {
+                        "trial_name": trial_name,
+                        "task_name": f"nvidia/{folder}{arm_suffix}_attempt1",
+                        "verifier_result": {"rewards": {"reward": 1.0 if is_with else 0.25}},
+                    }
+                ),
+                encoding="utf-8",
+            )
+        (job_dir / "result.json").write_text(
+            json.dumps(
+                {
+                    "n_total_trials": len(trial_names),
+                    "stats": {
+                        "n_trials": len(trial_names),
+                        "n_errors": 0,
+                        "evals": {
+                            agent: {
+                                "n_trials": len(trial_names),
+                                "n_errors": 0,
+                                "reward_stats": {"reward": {"1.0": trial_names}},
+                            }
+                        },
+                    },
+                }
+            ),
+            encoding="utf-8",
+        )
+        return True, ""
+
+    monkeypatch.setattr(runner, "_run_harbor", fake_run_harbor)
+
+    results = runner.run_harbor_eval(
+        skill_path=skill_dir,
+        agents=["opencode"],
+        output_dir=output_dir,
+        env_mode="docker",
+        n_attempts=2,
+        stop_on_pass=True,
+        agent_runtime_preflight=False,
+    )
+
+    assert results["execution_status"] == "succeeded"
+    # stop_on_pass passes physical task directory names to Harbor CLI --include-task-name
+    # With-skill passes on attempt 1 (1.0 >= 0.50), baseline runs 2 attempts per task (0.25 < 0.50)
+    assert seen_include_task_names == [
+        ["physical-case"],
+        ["zero-case"],
+        ["physical-case"],
+        ["physical-case"],
+        ["zero-case"],
+        ["zero-case"],
+    ]
+
+    agent_res = results["agents"]["opencode"]
+    assert set(agent_res["pass_at_k"]["with_skill"]["cases"]) == {"authored-entry", "0"}
+    assert set(agent_res["pass_at_k"]["without_skill"]["cases"]) == {"authored-entry", "0"}
+    assert agent_res["pass_at_k"]["with_skill"]["cases"]["authored-entry"]["best_score"] == 1.0
+    assert agent_res["pass_at_k"]["without_skill"]["cases"]["authored-entry"]["best_score"] == 0.25
+
+    run_dir = Path(results["run_dir"])
+    snapshot = json.loads((run_dir / "dataset_snapshot.json").read_text(encoding="utf-8"))
+    assert snapshot["dataset"] == [{"id": "authored-entry"}, {"id": "0"}]
+
+
 def test_stage_native_harbor_tasks_type_safety(tmp_path: Path) -> None:
     """Verify that native staging functions reject non-string arm_suffix values."""
     skill_dir = tmp_path / "target-skill"