diff --git a/.github/workflows/maint-80-langsmith-metrics-dashboard.yml b/.github/workflows/maint-80-langsmith-metrics-dashboard.yml
index 38c33c96e..769751061 100644
--- a/.github/workflows/maint-80-langsmith-metrics-dashboard.yml
+++ b/.github/workflows/maint-80-langsmith-metrics-dashboard.yml
@@ -155,6 +155,7 @@ jobs:
           set -uo pipefail
           REGISTRY=config/langsmith_fleet_registry.json
           mkdir -p .metrics-tmp/fleet
+          trusted_workflow_paths="$(jq -c '.trusted_artifact_workflow_paths // []' "$REGISTRY")"
           # Always start from an empty combined file so repos with no artifact
           # surface as "missing" (the registry-driven rollup never skips them).
           : > .metrics-tmp/fleet/combined-fleet.ndjson
@@ -178,7 +179,7 @@ jobs:
           # treated as "missing", never as a job failure.
           while IFS=$'\t' read -r repo artifact_name; do
             [ -z "$repo" ] && continue
-            echo "🔎 Resolving $artifact_name in $repo ..."
+            echo "🔎 Resolving $artifact_name or prefixed variant in $repo ..."
             artifacts_json=$(gh api --paginate --slurp --method GET \
               -H "Accept: application/vnd.github+json" \
               -H "X-GitHub-Api-Version: 2022-11-28" \
@@ -186,8 +187,41 @@ jobs:
               2>> .metrics-tmp/fleet/fleet_errors.log || echo "")
             artifact_id=$(printf '%s' "$artifacts_json" | jq -r \
               --arg artifact_name "$artifact_name" \
-              '[.[].artifacts[]? | select(.name == $artifact_name and .expired == false)] | sort_by(.created_at) | last | .id // empty' \
+              '[.[].artifacts[]?
+                | select(.expired == false)
+                | select(.name == $artifact_name)]
+              | sort_by(.created_at)
+              | last
+              | .id // empty' \
               2>> .metrics-tmp/fleet/fleet_errors.log || echo "")
+            if [ -z "$artifact_id" ] || [ "$artifact_id" = "null" ]; then
+              prefixed_candidates=$(printf '%s' "$artifacts_json" | jq -r \
+                --arg artifact_name "$artifact_name" \
+                '[.[].artifacts[]?
+                  | select(.expired == false)
+                  | select(.name != $artifact_name and (.name | endswith($artifact_name)))]
+                | sort_by(.created_at)
+                | reverse[]
+                | [.id, (.workflow_run.id // "")]
+                | @tsv' \
+                2>> .metrics-tmp/fleet/fleet_errors.log || echo "")
+              while IFS=$'\t' read -r candidate_id candidate_run_id; do
+                [ -z "$candidate_id" ] && continue
+                [ -z "$candidate_run_id" ] && continue
+                candidate_run_path=$(gh api --method GET \
+                  -H "Accept: application/vnd.github+json" \
+                  -H "X-GitHub-Api-Version: 2022-11-28" \
+                  "/repos/$repo/actions/runs/$candidate_run_id" \
+                  --jq '.path // ""' \
+                  2>> .metrics-tmp/fleet/fleet_errors.log || echo "")
+                if printf '%s' "$trusted_workflow_paths" | jq -e \
+                  --arg path "$candidate_run_path" \
+                  'index($path) != null' >/dev/null; then
+                  artifact_id="$candidate_id"
+                  break
+                fi
+              done <<< "$prefixed_candidates"
+            fi
             if [ -z "$artifact_id" ] || [ "$artifact_id" = "null" ]; then
               echo "  ↪ no current artifact for $repo (missing)"
               continue
diff --git a/.github/workflows/maint-81-langsmith-fleet-conformance.yml b/.github/workflows/maint-81-langsmith-fleet-conformance.yml
index 195ed2265..12949bea0 100644
--- a/.github/workflows/maint-81-langsmith-fleet-conformance.yml
+++ b/.github/workflows/maint-81-langsmith-fleet-conformance.yml
@@ -50,6 +50,9 @@ jobs:
             const registry = JSON.parse(
               fs.readFileSync('config/langsmith_fleet_registry.json', 'utf8')
             );
+            const trustedWorkflowPaths = new Set(
+              registry.trusted_artifact_workflow_paths || []
+            );
             fs.mkdirSync('.metrics-tmp/fleet', { recursive: true });
 
             for (const entry of registry.repos) {
@@ -62,11 +65,41 @@ jobs:
                 github.rest.actions.listArtifactsForRepo({
                   owner,
                   repo: repoName,
-                  name: entry.artifact_name,
                   per_page: 100,
                 })
               );
-              const artifact = artifacts.data.artifacts.find((item) => !item.expired);
+              const candidates = artifacts.data.artifacts.filter((item) =>
+                !item.expired &&
+                (item.name === entry.artifact_name || item.name.endsWith(entry.artifact_name))
+              );
+              const exactCandidates = candidates.filter(
+                (item) => item.name === entry.artifact_name
+              );
+              let artifact = exactCandidates.sort(
+                (left, right) => new Date(right.created_at) - new Date(left.created_at)
+              )[0];
+              if (!artifact) {
+                const suffixCandidates = candidates
+                  .filter((item) => item.name !== entry.artifact_name)
+                  .sort((left, right) => new Date(right.created_at) - new Date(left.created_at));
+                for (const candidate of suffixCandidates) {
+                  const runId = candidate.workflow_run && candidate.workflow_run.id;
+                  if (!runId) {
+                    continue;
+                  }
+                  const run = await withRetry(() =>
+                    github.rest.actions.getWorkflowRun({
+                      owner,
+                      repo: repoName,
+                      run_id: runId,
+                    })
+                  );
+                  if (trustedWorkflowPaths.has(run.data.path)) {
+                    artifact = candidate;
+                    break;
+                  }
+                }
+              }
               if (!artifact) {
                 core.warning(`No ${entry.artifact_name} artifact found for ${entry.repo}`);
                 continue;
diff --git a/.github/workflows/reusable-10-ci-python.yml b/.github/workflows/reusable-10-ci-python.yml
index a9634b496..4c5175493 100644
--- a/.github/workflows/reusable-10-ci-python.yml
+++ b/.github/workflows/reusable-10-ci-python.yml
@@ -1532,8 +1532,11 @@ jobs:
           ref: ${{ inputs.workflows_ref || 'main' }}
           path: .workflows-lib
           token: ${{ steps.app_token.outputs.token || github.token }}
+          persist-credentials: false
           sparse-checkout: |
             .github/actions/artifact-cache
+            config/langsmith_fleet_registry.json
+            scripts/ensure_langsmith_fleet_artifact.py
           sparse-checkout-cone-mode: false
 
       - name: Install uv
@@ -2451,6 +2454,62 @@ jobs:
           retention-days: 7
           overwrite: true
 
+      - name: Checkout Workflows LangSmith fleet helper
+        if: >-
+          ${{
+          always()
+          && !inputs.cache
+          && matrix.python-version == env.PRIMARY_PYTHON_VERSION
+          }}
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          repository: stranske/Workflows
+          # @main-only policy (issue #2346): `@main` is the single supported pin,
+          # so checking out the helper layer at `main` is intentional and correct
+          # - callers ride `@main`, hence main IS the pinned ref. The
+          # `workflows_ref` input is a vestige that defaults to 'main'; do not
+          # rely on it to pin helpers at a non-main ref (the helper layer is only
+          # validated as a unit on `main`).
+          ref: ${{ inputs.workflows_ref || 'main' }}
+          path: .workflows-lib
+          token: ${{ steps.app_token.outputs.token || github.token }}
+          persist-credentials: false
+          sparse-checkout: |
+            config/langsmith_fleet_registry.json
+            scripts/ensure_langsmith_fleet_artifact.py
+          sparse-checkout-cone-mode: false
+
+      - name: Ensure LangSmith fleet telemetry artifact
+        if: >-
+          ${{
+          always()
+          && matrix.python-version == env.PRIMARY_PYTHON_VERSION
+          }}
+        env:
+          PYTHON_VERSION: ${{ matrix.python-version }}
+        run: |
+          helper="${GITHUB_WORKSPACE}/.workflows-lib/scripts/ensure_langsmith_fleet_artifact.py"
+          registry="${GITHUB_WORKSPACE}/.workflows-lib/config/langsmith_fleet_registry.json"
+          if [ ! -f "$helper" ]; then
+            helper="${GITHUB_WORKSPACE}/scripts/ensure_langsmith_fleet_artifact.py"
+            registry="${GITHUB_WORKSPACE}/config/langsmith_fleet_registry.json"
+          fi
+          if [ ! -f "$helper" ]; then
+            echo "::warning::LangSmith fleet fallback helper is unavailable; skipping fallback artifact ensure."
+            exit 0
+          fi
+          python "$helper" \
+            --registry "$registry" \
+            --project-root "${PROJECT_ROOT}" \
+            --repository "${GITHUB_REPOSITORY}" \
+            --run-id "${GITHUB_RUN_ID}" \
+            --run-attempt "${GITHUB_RUN_ATTEMPT}" \
+            --workflow "${GITHUB_WORKFLOW}" \
+            --job "${GITHUB_JOB}" \
+            --python-version "${PYTHON_VERSION}" \
+            --sha "${GITHUB_SHA}" \
+            --event-name "${GITHUB_EVENT_NAME}"
+
       - name: Check LangSmith fleet telemetry artifact
         id: langsmith_fleet_artifact
         if: >-
@@ -2478,7 +2537,7 @@ jobs:
         continue-on-error: true
         uses: actions/upload-artifact@v7
         with:
-          name: ${{ inputs['artifact-prefix'] }}langsmith-fleet
+          name: ${{ inputs['artifact-prefix'] }}langsmith-fleet.ndjson
           path: ${{ env.PROJECT_ROOT }}/artifacts/langsmith/langsmith-fleet.ndjson
           if-no-files-found: warn
           retention-days: 90
diff --git a/config/langsmith_fleet_registry.json b/config/langsmith_fleet_registry.json
index 1b488cfe1..2f73dc73e 100644
--- a/config/langsmith_fleet_registry.json
+++ b/config/langsmith_fleet_registry.json
@@ -1,6 +1,11 @@
 {
   "schema_version": "langsmith-fleet-registry/v1",
   "stale_after_hours": 168,
+  "trusted_artifact_workflow_paths": [
+    ".github/workflows/pr-00-gate.yml",
+    ".github/workflows/selftest-reusable-ci.yml",
+    ".github/workflows/maint-62-integration-consumer.yml"
+  ],
   "repos": [
     {
       "repo": "stranske/Workflows",
diff --git a/scripts/ensure_langsmith_fleet_artifact.py b/scripts/ensure_langsmith_fleet_artifact.py
new file mode 100644
index 000000000..7af724a9d
--- /dev/null
+++ b/scripts/ensure_langsmith_fleet_artifact.py
@@ -0,0 +1,313 @@
+#!/usr/bin/env python3
+"""Ensure reusable CI has a LangSmith fleet artifact to upload.
+
+The normal producer remains repo-local instrumentation. This helper only writes
+an explicit CI fallback row when an implemented registry repo produced no
+``artifacts/langsmith/langsmith-fleet.ndjson`` file, so artifact distribution can
+still be diagnosed from failed or partial CI runs.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+try:
+    from datetime import UTC
+except ImportError:  # pragma: no cover - Python < 3.11 compatibility
+    from datetime import timezone
+
+    UTC = timezone.utc  # noqa: UP017 - fallback for Python < 3.11
+
+SCHEMA_VERSION = "langsmith-fleet/v1"
+ARTIFACT_NAME = "langsmith-fleet.ndjson"
+FALLBACK_ERROR_CATEGORY = "ci_fleet_artifact_missing"
+ELIGIBLE_ROLLOUT_STATUSES = {"implemented", "implemented-followup-open"}
+
+
+def load_registry(path: Path) -> dict[str, Any]:
+    """Load the fleet registry JSON."""
+    with path.open("r", encoding="utf-8") as handle:
+        data = json.load(handle)
+    if not isinstance(data, dict):
+        raise ValueError("registry must be a JSON object")
+    repos = data.get("repos")
+    if not isinstance(repos, list):
+        raise ValueError("registry repos must be a list")
+    return data
+
+
+def _artifact_has_records(path: Path) -> bool:
+    """Return whether an artifact file already contains at least one row."""
+    if not path.exists() or not path.is_file():
+        return False
+    try:
+        return any(line.strip() for line in path.read_text(encoding="utf-8").splitlines())
+    except OSError:
+        return False
+
+
+def _has_langsmith_artifact_contract(entry: dict[str, Any]) -> bool:
+    """Return whether a registry entry has the fields needed for fallback rows."""
+    operations = entry.get("operations")
+    required_domain_fields = entry.get("required_domain_fields")
+    return (
+        str(entry.get("artifact_name", "")).strip() == ARTIFACT_NAME
+        and bool(str(entry.get("surface", "")).strip())
+        and bool(str(entry.get("issue", "")).strip())
+        and isinstance(operations, list)
+        and any(str(operation).strip() for operation in operations)
+        and isinstance(required_domain_fields, list)
+        and all(str(field).strip() for field in required_domain_fields)
+    )
+
+
+def _entry_for_repo(
+    registry: dict[str, Any], repository: str
+) -> tuple[dict[str, Any] | None, str | None]:
+    """Find one unambiguous fallback-eligible LangSmith registry entry."""
+    normalized = repository.strip()
+    repo_entries = [
+        entry
+        for entry in registry.get("repos", [])
+        if isinstance(entry, dict) and str(entry.get("repo", "")).strip() == normalized
+    ]
+    if not repo_entries:
+        return None, "repository_not_in_registry"
+    contract_entries = [entry for entry in repo_entries if _has_langsmith_artifact_contract(entry)]
+    if not contract_entries:
+        return None, "repository_missing_langsmith_artifact_contract"
+    eligible_entries = [
+        entry
+        for entry in contract_entries
+        if str(entry.get("rollout_status") or "").strip() in ELIGIBLE_ROLLOUT_STATUSES
+    ]
+    if len(eligible_entries) == 1:
+        return eligible_entries[0], None
+    if len(eligible_entries) > 1:
+        return None, "repository_langsmith_artifact_contract_ambiguous"
+    if len(contract_entries) == 1:
+        return contract_entries[0], None
+    return None, "repository_langsmith_artifact_contract_ambiguous"
+
+
+def _domain_fallback_value(field: str, *, now: datetime) -> Any:
+    """Build a schema-friendly placeholder for a required domain field."""
+    lowered = field.lower()
+    if lowered in {"as_of_date"} or lowered.endswith("_date"):
+        return now.date().isoformat()
+    if lowered in {"seed", "row_count", "tool_call_count"} or lowered.endswith("_count"):
+        return 0
+    if lowered.endswith("_score") or lowered.endswith("_delta"):
+        return 0
+    if lowered.endswith("_status") or lowered in {"fallback_state", "result"}:
+        return "ci_fallback_no_records"
+    if lowered.endswith("_hash"):
+        return "ref:ci-fallback-no-records"
+    if lowered.endswith("_id"):
+        return "ci-fallback-no-records"
+    return "ci-fallback-no-records"
+
+
+def build_fallback_record(
+    entry: dict[str, Any],
+    *,
+    repository: str,
+    run_id: str,
+    run_attempt: str,
+    workflow: str,
+    job: str,
+    python_version: str,
+    sha: str,
+    event_name: str,
+    now: datetime,
+) -> dict[str, Any]:
+    """Build one registry-valid row that identifies missing CI producer output."""
+    operations = entry.get("operations") if isinstance(entry.get("operations"), list) else []
+    operation = str(operations[0]).strip() if operations else "ci-fallback"
+    required_domain_fields = (
+        entry.get("required_domain_fields")
+        if isinstance(entry.get("required_domain_fields"), list)
+        else []
+    )
+    domain = {
+        str(field): _domain_fallback_value(str(field), now=now)
+        for field in required_domain_fields
+        if str(field).strip()
+    }
+    domain.update(
+        {
+            "workflow": workflow or "unknown",
+            "job": job or "unknown",
+            "python_version": python_version or "unknown",
+            "run_attempt": run_attempt or "1",
+            "event_name": event_name or "unknown",
+            "fallback_reason": FALLBACK_ERROR_CATEGORY,
+            "result": "no_ci_fleet_records",
+        }
+    )
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "repo": repository,
+        "surface": str(entry.get("surface") or "unknown"),
+        "operation": operation,
+        "run_id": f"github-actions:{run_id or 'unknown'}:{run_attempt or '1'}:langsmith-fleet",
+        "status": "error",
+        "github_issue": str(entry.get("issue") or ""),
+        "recorded_at": now.isoformat().replace("+00:00", "Z"),
+        "input_hash": f"ref:{sha}" if sha else "ref:unknown",
+        "output_hash": "artifact:langsmith-fleet-fallback",
+        "artifact_ref": f"artifact:{ARTIFACT_NAME}",
+        "error_category": FALLBACK_ERROR_CATEGORY,
+        "domain": domain,
+    }
+
+
+def ensure_artifact(
+    *,
+    artifact_path: Path,
+    registry_path: Path,
+    repository: str,
+    run_id: str,
+    run_attempt: str,
+    workflow: str,
+    job: str,
+    python_version: str,
+    sha: str,
+    event_name: str,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Create a fallback artifact if an eligible repo produced no records."""
+    if _artifact_has_records(artifact_path):
+        return {
+            "status": "existing",
+            "artifact_path": str(artifact_path),
+            "repository": repository,
+        }
+
+    registry = load_registry(registry_path)
+    entry, skipped_reason = _entry_for_repo(registry, repository)
+    if entry is None:
+        return {
+            "status": "skipped",
+            "reason": skipped_reason or "repository_not_in_registry",
+            "artifact_path": str(artifact_path),
+            "repository": repository,
+        }
+
+    rollout_status = str(entry.get("rollout_status") or "").strip()
+    if rollout_status not in ELIGIBLE_ROLLOUT_STATUSES:
+        return {
+            "status": "skipped",
+            "reason": f"rollout_status_{rollout_status or 'missing'}",
+            "artifact_path": str(artifact_path),
+            "repository": repository,
+        }
+
+    timestamp = now or datetime.now(UTC)
+    record = build_fallback_record(
+        entry,
+        repository=repository,
+        run_id=run_id,
+        run_attempt=run_attempt,
+        workflow=workflow,
+        job=job,
+        python_version=python_version,
+        sha=sha,
+        event_name=event_name,
+        now=timestamp,
+    )
+    artifact_path.parent.mkdir(parents=True, exist_ok=True)
+    artifact_path.write_text(
+        json.dumps(record, sort_keys=True, separators=(",", ":")) + "\n",
+        encoding="utf-8",
+    )
+    return {
+        "status": "created",
+        "reason": FALLBACK_ERROR_CATEGORY,
+        "artifact_path": str(artifact_path),
+        "repository": repository,
+        "surface": record["surface"],
+        "operation": record["operation"],
+    }
+
+
+def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
+    """Parse CLI arguments and environment defaults."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--artifact-path",
+        type=Path,
+        default=None,
+        help="Target langsmith-fleet.ndjson path.",
+    )
+    parser.add_argument(
+        "--project-root",
+        type=Path,
+        default=Path(os.environ.get("PROJECT_ROOT") or "."),
+        help="Project root used when --artifact-path is omitted.",
+    )
+    parser.add_argument(
+        "--registry",
+        type=Path,
+        default=Path("config/langsmith_fleet_registry.json"),
+        help="Workflows LangSmith fleet registry path.",
+    )
+    parser.add_argument("--repository", default=os.environ.get("GITHUB_REPOSITORY", ""))
+    parser.add_argument("--run-id", default=os.environ.get("GITHUB_RUN_ID", ""))
+    parser.add_argument("--run-attempt", default=os.environ.get("GITHUB_RUN_ATTEMPT", "1"))
+    parser.add_argument("--workflow", default=os.environ.get("GITHUB_WORKFLOW", ""))
+    parser.add_argument("--job", default=os.environ.get("GITHUB_JOB", ""))
+    parser.add_argument("--python-version", default=os.environ.get("PYTHON_VERSION", ""))
+    parser.add_argument("--sha", default=os.environ.get("GITHUB_SHA", ""))
+    parser.add_argument("--event-name", default=os.environ.get("GITHUB_EVENT_NAME", ""))
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    """CLI entry point used by reusable CI."""
+    args = parse_args(argv)
+    artifact_path = args.artifact_path or (
+        args.project_root / "artifacts" / "langsmith" / ARTIFACT_NAME
+    )
+    try:
+        result = ensure_artifact(
+            artifact_path=artifact_path,
+            registry_path=args.registry,
+            repository=args.repository,
+            run_id=args.run_id,
+            run_attempt=args.run_attempt,
+            workflow=args.workflow,
+            job=args.job,
+            python_version=args.python_version,
+            sha=args.sha,
+            event_name=args.event_name,
+        )
+    except Exception as exc:  # pragma: no cover - defensive CI fail-open path
+        result = {
+            "status": "skipped",
+            "reason": "ensure_failed",
+            "error": str(exc),
+            "artifact_path": str(artifact_path),
+            "repository": args.repository,
+        }
+        print(
+            f"::warning::LangSmith fleet fallback artifact ensure failed: {exc}",
+            file=sys.stderr,
+        )
+    print(json.dumps(result, sort_keys=True))
+    if result.get("status") == "created":
+        print(
+            "::notice::Created LangSmith fleet fallback artifact because no repo-produced "
+            f"records were present: {artifact_path}"
+        )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/scripts/test_ensure_langsmith_fleet_artifact.py b/tests/scripts/test_ensure_langsmith_fleet_artifact.py
new file mode 100644
index 000000000..d0b974c9a
--- /dev/null
+++ b/tests/scripts/test_ensure_langsmith_fleet_artifact.py
@@ -0,0 +1,190 @@
+import json
+from datetime import datetime
+from pathlib import Path
+
+from scripts import ensure_langsmith_fleet_artifact, langsmith_fleet
+
+try:
+    from datetime import UTC
+except ImportError:  # pragma: no cover - Python < 3.11 compatibility
+    from datetime import timezone
+
+    UTC = timezone.utc  # noqa: UP017 - fallback for Python < 3.11
+
+ROOT = Path(__file__).resolve().parents[2]
+REGISTRY = ROOT / "config" / "langsmith_fleet_registry.json"
+
+
+def test_ensure_artifact_writes_valid_error_record_for_implemented_repo(tmp_path: Path) -> None:
+    artifact = tmp_path / "artifacts" / "langsmith" / "langsmith-fleet.ndjson"
+    now = datetime(2026, 6, 24, 2, 30, tzinfo=UTC)
+
+    result = ensure_langsmith_fleet_artifact.ensure_artifact(
+        artifact_path=artifact,
+        registry_path=REGISTRY,
+        repository="stranske/Pension-Data",
+        run_id="28068938440",
+        run_attempt="1",
+        workflow="CI",
+        job="tests",
+        python_version="3.12",
+        sha="abc123",
+        event_name="push",
+        now=now,
+    )
+
+    assert result["status"] == "created"
+    record = json.loads(artifact.read_text(encoding="utf-8"))
+    assert record["schema_version"] == langsmith_fleet.SCHEMA_VERSION
+    assert record["repo"] == "stranske/Pension-Data"
+    assert record["surface"] == "nl-to-sql"
+    assert record["operation"] == "sql-generation"
+    assert record["status"] == "error"
+    assert record["error_category"] == "ci_fleet_artifact_missing"
+    assert record["run_id"] == "github-actions:28068938440:1:langsmith-fleet"
+    assert record["recorded_at"] == "2026-06-24T02:30:00Z"
+    assert record["input_hash"] == "ref:abc123"
+    assert record["domain"]["query_category"] == "ci-fallback-no-records"
+    assert record["domain"]["sql_validation_status"] == "ci_fallback_no_records"
+    assert record["domain"]["read_only_status"] == "ci_fallback_no_records"
+    assert record["domain"]["row_count"] == 0
+    assert record["domain"]["fallback_reason"] == "ci_fleet_artifact_missing"
+
+    registry = langsmith_fleet.load_registry(REGISTRY)
+    schema = langsmith_fleet.load_record_schema()
+    assert langsmith_fleet.validate_record(record, registry=registry, schema=schema) == []
+
+
+def test_ensure_artifact_preserves_existing_repo_records(tmp_path: Path) -> None:
+    artifact = tmp_path / "artifacts" / "langsmith" / "langsmith-fleet.ndjson"
+    artifact.parent.mkdir(parents=True)
+    original = '{"schema_version":"langsmith-fleet/v1","repo":"stranske/Pension-Data"}\n'
+    artifact.write_text(original, encoding="utf-8")
+
+    result = ensure_langsmith_fleet_artifact.ensure_artifact(
+        artifact_path=artifact,
+        registry_path=REGISTRY,
+        repository="stranske/Pension-Data",
+        run_id="28068938440",
+        run_attempt="1",
+        workflow="CI",
+        job="tests",
+        python_version="3.12",
+        sha="abc123",
+        event_name="push",
+    )
+
+    assert result["status"] == "existing"
+    assert artifact.read_text(encoding="utf-8") == original
+
+
+def test_ensure_artifact_skips_repos_without_implemented_artifact_rollout(tmp_path: Path) -> None:
+    artifact = tmp_path / "artifacts" / "langsmith" / "langsmith-fleet.ndjson"
+
+    result = ensure_langsmith_fleet_artifact.ensure_artifact(
+        artifact_path=artifact,
+        registry_path=REGISTRY,
+        repository="stranske/Travel-Plan-Permission",
+        run_id="28068938440",
+        run_attempt="1",
+        workflow="CI",
+        job="tests",
+        python_version="3.12",
+        sha="abc123",
+        event_name="push",
+    )
+
+    assert result == {
+        "status": "skipped",
+        "reason": "rollout_status_covered-via-langsmith-direct",
+        "artifact_path": str(artifact),
+        "repository": "stranske/Travel-Plan-Permission",
+    }
+    assert not artifact.exists()
+
+
+def test_ensure_artifact_skips_ambiguous_registry_contract(tmp_path: Path) -> None:
+    artifact = tmp_path / "artifacts" / "langsmith" / "langsmith-fleet.ndjson"
+    registry_path = tmp_path / "registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "repos": [
+                    {
+                        "repo": "stranske/Pension-Data",
+                        "issue": "stranske/Pension-Data#445",
+                        "surface": "nl-to-sql",
+                        "operations": ["sql-generation"],
+                        "artifact_name": "langsmith-fleet.ndjson",
+                        "rollout_status": "implemented",
+                        "required_domain_fields": ["query_category"],
+                    },
+                    {
+                        "repo": "stranske/Pension-Data",
+                        "issue": "stranske/Pension-Data#446",
+                        "surface": "benefits-summary",
+                        "operations": ["summary-generation"],
+                        "artifact_name": "langsmith-fleet.ndjson",
+                        "rollout_status": "implemented",
+                        "required_domain_fields": ["summary_status"],
+                    },
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    result = ensure_langsmith_fleet_artifact.ensure_artifact(
+        artifact_path=artifact,
+        registry_path=registry_path,
+        repository="stranske/Pension-Data",
+        run_id="28068938440",
+        run_attempt="1",
+        workflow="CI",
+        job="tests",
+        python_version="3.12",
+        sha="abc123",
+        event_name="push",
+    )
+
+    assert result == {
+        "status": "skipped",
+        "reason": "repository_langsmith_artifact_contract_ambiguous",
+        "artifact_path": str(artifact),
+        "repository": "stranske/Pension-Data",
+    }
+    assert not artifact.exists()
+
+
+def test_main_notice_uses_github_actions_annotation_prefix(tmp_path: Path, capsys) -> None:
+    artifact = tmp_path / "artifacts" / "langsmith" / "langsmith-fleet.ndjson"
+
+    rc = ensure_langsmith_fleet_artifact.main(
+        [
+            "--artifact-path",
+            str(artifact),
+            "--registry",
+            str(REGISTRY),
+            "--repository",
+            "stranske/Pension-Data",
+            "--run-id",
+            "28068938440",
+            "--run-attempt",
+            "1",
+            "--workflow",
+            "CI",
+            "--job",
+            "tests",
+            "--python-version",
+            "3.12",
+            "--sha",
+            "abc123",
+            "--event-name",
+            "push",
+        ]
+    )
+
+    captured = capsys.readouterr()
+    assert rc == 0
+    assert "::notice::Created LangSmith fleet fallback artifact" in captured.out
+    assert "::notice ::" not in captured.out
diff --git a/tests/workflows/test_langsmith_fleet_conformance_workflow.py b/tests/workflows/test_langsmith_fleet_conformance_workflow.py
new file mode 100644
index 000000000..e5a3a5e06
--- /dev/null
+++ b/tests/workflows/test_langsmith_fleet_conformance_workflow.py
@@ -0,0 +1,18 @@
+from pathlib import Path
+
+WORKFLOW = Path(".github/workflows/maint-81-langsmith-fleet-conformance.yml")
+
+
+def test_conformance_download_accepts_prefixed_fleet_artifacts() -> None:
+    source = WORKFLOW.read_text(encoding="utf-8")
+
+    assert source.count("listArtifactsForRepo") == 1
+    assert source.count("name: entry.artifact_name") == 0
+    assert (
+        source.count("item.name === entry.artifact_name || item.name.endsWith(entry.artifact_name)")
+        == 1
+    )
+    assert source.count("const exactCandidates = candidates.filter") == 1
+    assert source.count("trusted_artifact_workflow_paths") == 1
+    assert source.count("github.rest.actions.getWorkflowRun") == 1
+    assert source.count("trustedWorkflowPaths.has(run.data.path)") == 1
diff --git a/tests/workflows/test_langsmith_metrics_dashboard.py b/tests/workflows/test_langsmith_metrics_dashboard.py
index 8f9bd47ce..3aac7f20a 100644
--- a/tests/workflows/test_langsmith_metrics_dashboard.py
+++ b/tests/workflows/test_langsmith_metrics_dashboard.py
@@ -10,11 +10,25 @@ def test_fleet_artifact_lookup_does_not_mix_slurp_with_jq() -> None:
 
     assert "--paginate --slurp --method GET" in source
     assert "--arg artifact_name" in source
-    assert "select(.name == $artifact_name and .expired == false)" in source
+    assert "select(.name == $artifact_name)" in source
+    assert "select(.name != $artifact_name and (.name | endswith($artifact_name)))" in source
+    assert "trusted_artifact_workflow_paths" in source
+    assert '"/repos/$repo/actions/runs/$candidate_run_id"' in source
+    assert "'index($path) != null'" in source
     assert "--slurp --method GET \\\n" in source
     assert '--jq "[.[].artifacts' not in source
 
 
+def test_fleet_registry_declares_trusted_artifact_workflows() -> None:
+    registry = json.loads(FLEET_REGISTRY.read_text(encoding="utf-8"))
+
+    assert registry["trusted_artifact_workflow_paths"] == [
+        ".github/workflows/pr-00-gate.yml",
+        ".github/workflows/selftest-reusable-ci.yml",
+        ".github/workflows/maint-62-integration-consumer.yml",
+    ]
+
+
 def test_dashboard_issue_uses_existing_labels() -> None:
     source = WORKFLOW.read_text(encoding="utf-8")
 
diff --git a/tests/workflows/test_reusable_ci_workflow.py b/tests/workflows/test_reusable_ci_workflow.py
index 50214da0a..174f37d29 100644
--- a/tests/workflows/test_reusable_ci_workflow.py
+++ b/tests/workflows/test_reusable_ci_workflow.py
@@ -208,6 +208,45 @@ def _step(name: str) -> dict:
     primary_step = _step("Resolve primary python version")
     assert _normalize_expr(primary_step["if"]) == "${{always()}}"
 
+    workflows_helper_step = _step("Checkout Workflows artifact cache action")
+    assert workflows_helper_step["with"]["persist-credentials"] is False
+    helper_sparse_checkout = workflows_helper_step["with"]["sparse-checkout"]
+    assert ".github/actions/artifact-cache" in helper_sparse_checkout
+    assert "config/langsmith_fleet_registry.json" in helper_sparse_checkout
+    assert "scripts/ensure_langsmith_fleet_artifact.py" in helper_sparse_checkout
+
+    langsmith_helper_step = _step("Checkout Workflows LangSmith fleet helper")
+    assert (
+        langsmith_helper_step["uses"] == "actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0"
+    )
+    assert langsmith_helper_step["with"]["persist-credentials"] is False
+    assert _normalize_expr(langsmith_helper_step["if"]) == (
+        "${{always()&&!inputs.cache&&matrix.python-version==env.PRIMARY_PYTHON_VERSION}}"
+    )
+    assert (
+        "config/langsmith_fleet_registry.json" in langsmith_helper_step["with"]["sparse-checkout"]
+    )
+    assert (
+        "scripts/ensure_langsmith_fleet_artifact.py"
+        in langsmith_helper_step["with"]["sparse-checkout"]
+    )
+
+    langsmith_ensure_step = _step("Ensure LangSmith fleet telemetry artifact")
+    assert _normalize_expr(langsmith_ensure_step["if"]) == (
+        "${{always()&&matrix.python-version==env.PRIMARY_PYTHON_VERSION}}"
+    )
+    assert "scripts/ensure_langsmith_fleet_artifact.py" in langsmith_ensure_step["run"]
+    assert (
+        ".workflows-lib/scripts/ensure_langsmith_fleet_artifact.py" in langsmith_ensure_step["run"]
+    )
+    assert "fallback helper is unavailable" in langsmith_ensure_step["run"]
+    assert (
+        "::warning::LangSmith fleet fallback helper is unavailable" in langsmith_ensure_step["run"]
+    )
+    assert "--registry" in langsmith_ensure_step["run"]
+    assert "--project-root" in langsmith_ensure_step["run"]
+    assert "--repository" in langsmith_ensure_step["run"]
+
     langsmith_check_step = _step("Check LangSmith fleet telemetry artifact")
     assert langsmith_check_step["id"] == "langsmith_fleet_artifact"
     assert "artifacts/langsmith/langsmith-fleet.ndjson" in langsmith_check_step["run"]
@@ -217,7 +256,8 @@ def _step(name: str) -> dict:
     assert langsmith_upload_step["uses"] == "actions/upload-artifact@v7"
     assert langsmith_upload_step["continue-on-error"] is True
     assert (
-        langsmith_upload_step["with"]["name"] == "${{ inputs['artifact-prefix'] }}langsmith-fleet"
+        langsmith_upload_step["with"]["name"]
+        == "${{ inputs['artifact-prefix'] }}langsmith-fleet.ndjson"
     )
     assert (
         langsmith_upload_step["with"]["path"]
@@ -236,6 +276,12 @@ def _step(name: str) -> dict:
     assert step_names.index("Check LangSmith fleet telemetry artifact") > step_names.index(
         "Upload coverage trend history artifact"
     )
+    assert step_names.index("Ensure LangSmith fleet telemetry artifact") > step_names.index(
+        "Upload coverage trend history artifact"
+    )
+    assert step_names.index("Check LangSmith fleet telemetry artifact") == (
+        step_names.index("Ensure LangSmith fleet telemetry artifact") + 1
+    )
     assert step_names.index("Upload LangSmith fleet telemetry artifact") == (
         step_names.index("Check LangSmith fleet telemetry artifact") + 1
     )