diff --git a/.github/workflows/codeql-scan-dispatch.yml b/.github/workflows/codeql-scan-dispatch.yml
index 04656e5b34..b89cd1a85f 100644
--- a/.github/workflows/codeql-scan-dispatch.yml
+++ b/.github/workflows/codeql-scan-dispatch.yml
@@ -439,7 +439,7 @@ jobs:
     permissions:
       actions: read
       contents: read
-      security-events: read
+      security-events: write # Same-target SARIF POST; cross-repo uses a separate scoped token.
       id-token: write
       statuses: write # Required for downscoped OIDC status publication.
     strategy:
@@ -556,7 +556,10 @@ jobs:
           python3 -c "import ast; ast.parse(open('$RUNNER_TEMP/codeql_sarif_gate.py').read())"
           gh api "repos/ContextualWisdomLab/.github/contents/scripts/ci/codeql_ghas_configuration_identity.py?ref=${WORKFLOW_SHA}" \
             --jq .content | base64 --decode >"$RUNNER_TEMP/codeql_ghas_configuration_identity.py"
-          python3 -c "import ast; ast.parse(open('$RUNNER_TEMP/codeql_ghas_configuration_identity.py').read())"
+          python3 -I -c "import ast; ast.parse(open('$RUNNER_TEMP/codeql_ghas_configuration_identity.py').read())"
+          gh api "repos/ContextualWisdomLab/.github/contents/scripts/ci/codeql_sarif_publication.py?ref=${WORKFLOW_SHA}" \
+            --jq .content | base64 --decode >"$RUNNER_TEMP/codeql_sarif_publication.py"
+          python3 -I -c "import ast; ast.parse(open('$RUNNER_TEMP/codeql_sarif_publication.py').read())"
 
       - name: Materialize pull request head for CodeQL scan
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
@@ -659,9 +662,46 @@ jobs:
           echo "::error::no configured credential can read target CodeQL analyses; GHAS configuration identity cannot be proven."
           exit 1
 
+      - name: Mint target-scoped Noema SARIF publication token
+        id: noema_sarif_token
+        if: always() && steps.live_metadata.outcome == 'success' && needs.validate-dispatch.outputs.target_repository != github.repository && steps.noema_analysis_config.outputs.available == 'true'
+        uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0
+        with:
+          client-id: ${{ vars.NOEMA_GITHUB_APP_CLIENT_ID }}
+          private-key: ${{ secrets.NOEMA_GITHUB_APP_PRIVATE_KEY }}
+          owner: ContextualWisdomLab
+          repositories: ${{ steps.noema_analysis_config.outputs.repository }}
+          permission-security-events: write
+
+      - name: Publish and verify CodeQL SARIF in target code scanning
+        id: sarif_publication
+        if: always() && steps.live_metadata.outcome == 'success' && (steps.gate.outcome == 'success' || steps.gate.outcome == 'failure')
+        env:
+          GH_TOKEN: ${{ needs.validate-dispatch.outputs.target_repository == github.repository && github.token || steps.noema_sarif_token.outputs.token }}
+          REPOSITORY: ${{ needs.validate-dispatch.outputs.target_repository }}
+          PR_NUMBER: ${{ needs.validate-dispatch.outputs.pr_number }}
+          BASE_SHA: ${{ needs.validate-dispatch.outputs.base_sha }}
+          HEAD_SHA: ${{ needs.validate-dispatch.outputs.head_sha }}
+          LANGUAGE: ${{ matrix.language }}
+          WORKFLOW_SHA: ${{ github.workflow_sha }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+        run: |
+          set -euo pipefail
+          # Outside checkout and RUNNER_TEMP: original evidence must survive
+          # checkout cleanup, job posts and upload/processing failure. No GC here.
+          custody="$(dirname "$RUNNER_WORKSPACE")/.cwl-codeql-sarif/${RUN_ID}/${RUN_ATTEMPT}/${LANGUAGE}"
+          echo "::add-mask::$custody"
+          if [ "$(git rev-parse HEAD)" != "$HEAD_SHA" ]; then
+            echo "::error::CodeQL SARIF checkout does not match the admitted head."
+            exit 1
+          fi
+          python3 -I "$RUNNER_TEMP/codeql_sarif_publication.py" publish --identity-env \
+            --source codeql-results-dispatch --custody "$custody"
+
       - name: Verify GHAS base/head CodeQL configuration identity
         id: ghas_configuration_identity
-        if: steps.gate.outcome == 'success'
+        if: steps.gate.outcome == 'success' && steps.sarif_publication.outcome == 'success'
         env:
           GH_TOKEN: ${{ steps.ghas_analysis_token.outputs.token }}
           TARGET_REPOSITORY: ${{ needs.validate-dispatch.outputs.target_repository }}
@@ -693,13 +733,23 @@ jobs:
 
       - name: Preserve CodeQL SARIF evidence
         id: sarif_upload
-        if: always() && hashFiles('codeql-results-dispatch/**/*.sarif') != ''
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        with:
-          name: codeql-dispatch-${{ matrix.language }}-${{ github.run_id }}-${{ github.run_attempt }}
-          path: codeql-results-dispatch
-          if-no-files-found: error
-          retention-days: 7
+        if: always() && steps.sarif_publication.outcome == 'success'
+        env:
+          GH_TOKEN: ${{ needs.validate-dispatch.outputs.target_repository == github.repository && github.token || steps.noema_sarif_token.outputs.token }}
+          REPOSITORY: ${{ needs.validate-dispatch.outputs.target_repository }}
+          PR_NUMBER: ${{ needs.validate-dispatch.outputs.pr_number }}
+          BASE_SHA: ${{ needs.validate-dispatch.outputs.base_sha }}
+          HEAD_SHA: ${{ needs.validate-dispatch.outputs.head_sha }}
+          LANGUAGE: ${{ matrix.language }}
+          WORKFLOW_SHA: ${{ github.workflow_sha }}
+          RUN_ID: ${{ github.run_id }}
+          RUN_ATTEMPT: ${{ github.run_attempt }}
+        run: |
+          set -euo pipefail
+          # Preserve the established receiver step name, but actually rehash
+          # original/upload/readback bytes and recheck the accepted analysis.
+          custody="$(dirname "$RUNNER_WORKSPACE")/.cwl-codeql-sarif/${RUN_ID}/${RUN_ATTEMPT}/${LANGUAGE}"
+          python3 -I "$RUNNER_TEMP/codeql_sarif_publication.py" verify-custody --identity-env --custody "$custody"
 
       - name: Mint target-scoped Noema CodeQL status token
         id: noema_status_token
@@ -850,7 +900,7 @@ jobs:
           fi
 
           if [ "$GATE_OUTCOME" = "success" ]; then
-            echo "::notice::Could not publish the CodeQL dispatch status after all configured credentials failed. The exact completed scan and preserved SARIF artifact remain the authenticated fallback evidence."
+            echo "::notice::Could not publish the CodeQL dispatch status after all configured credentials failed. The exact completed scan, durable custody receipt and processed target SARIF remain the authenticated fallback evidence."
             exit 0
           fi
 
@@ -872,8 +922,19 @@ jobs:
     permissions:
       actions: write
       contents: read
+      security-events: read # Readback only; never reuse the lifecycle token to upload.
       id-token: write
     steps:
+      - name: Fetch pinned SARIF settlement verifier
+        env:
+          GH_TOKEN: ${{ github.token }}
+          WORKFLOW_SHA: ${{ github.workflow_sha }}
+        run: |
+          set -euo pipefail
+          gh api "repos/ContextualWisdomLab/.github/contents/scripts/ci/codeql_sarif_publication.py?ref=${WORKFLOW_SHA}" \
+            --jq .content | base64 --decode >"$RUNNER_TEMP/codeql_sarif_publication.py"
+          python3 -I -c "import ast; ast.parse(open('$RUNNER_TEMP/codeql_sarif_publication.py').read())"
+
       - name: Exchange OpenCode app token for run settlement
         id: target_app_token
         env:
@@ -966,8 +1027,21 @@ jobs:
           repositories: ${{ steps.noema_settlement_config.outputs.repository }}
           permission-actions: write
 
+      - name: Mint target-scoped Noema SARIF settlement reader
+        id: noema_sarif_reader
+        if: needs.validate-dispatch.outputs.target_repository != github.repository && steps.noema_settlement_config.outputs.available == 'true'
+        uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0
+        with:
+          client-id: ${{ vars.NOEMA_GITHUB_APP_CLIENT_ID }}
+          private-key: ${{ secrets.NOEMA_GITHUB_APP_PRIVATE_KEY }}
+          owner: ContextualWisdomLab
+          repositories: ${{ steps.noema_settlement_config.outputs.repository }}
+          permission-security-events: read
+
       - name: Settle exact CodeQL required run
         env:
+          SARIF_READ_TOKEN: ${{ needs.validate-dispatch.outputs.target_repository == github.repository && github.token || steps.noema_sarif_reader.outputs.token }}
+          WORKFLOW_SHA: ${{ github.workflow_sha }}
           NOEMA_WAKE_TOKEN: ${{ steps.noema_settlement_token.outputs.token || '' }}
           TARGET_APP_WAKE_TOKEN: ${{ steps.target_app_token.outputs.token || '' }}
           PR_REVIEW_MERGE_WAKE_TOKEN: ${{ secrets.PR_REVIEW_MERGE_TOKEN || '' }}
@@ -1094,17 +1168,16 @@ jobs:
             exit 1
           fi
 
-          if ! handler_job_pages="$(GH_TOKEN="$HANDLER_READ_TOKEN" gh api --paginate "repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/jobs?per_page=100" | jq -s .)" ||
-            ! handler_artifact_pages="$(GH_TOKEN="$HANDLER_READ_TOKEN" gh api --paginate "repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/artifacts?per_page=100" | jq -s .)"; then
+          if ! handler_job_pages="$(GH_TOKEN="$HANDLER_READ_TOKEN" gh api --paginate "repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/jobs?per_page=100" | jq -s .)"; then
             echo "::error::CodeQL settlement could not read exact handler evidence."
             exit 1
           fi
           handler_jobs="$(printf '%s' "$handler_job_pages" | jq -c '[.[] | .jobs[]?]')"
-          handler_artifacts="$(printf '%s' "$handler_artifact_pages" | jq -c '[.[] | .artifacts[]?]')"
           while IFS= read -r required_job; do
             language="$(printf '%s' "$required_job" | jq -r '.language')"
             expected_job_name="CodeQL dispatch scan (${language})"
-            expected_artifact_name="codeql-dispatch-${language}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+            # A successful preserved-evidence step now means real target SARIF
+            # processing plus durable original-byte custody, not CreateArtifact.
             handler_job_count="$(printf '%s' "$handler_jobs" | jq --arg name "$expected_job_name" --argjson attempt "$GITHUB_RUN_ATTEMPT" '
               [.[] | select(
                 .name == $name
@@ -1115,13 +1188,30 @@ jobs:
                 and ([.steps[]? | select(.name == "Preserve CodeQL SARIF evidence" and .conclusion == "success")] | length) == 1
               )] | length
             ')"
-            handler_artifact_count="$(printf '%s' "$handler_artifacts" | jq --arg name "$expected_artifact_name" '
-              [.[] | select(.name == $name and (.expired == false) and (.size_in_bytes > 0))] | length
-            ')"
-            if [ "$handler_job_count" -ne 1 ] || [ "$handler_artifact_count" -ne 1 ]; then
+            if [ "$handler_job_count" -ne 1 ]; then
               echo "::error::CodeQL settlement rejected incomplete handler gate or SARIF evidence for ${language}."
               exit 1
             fi
+            if ! handler_job_id="$(printf '%s' "$handler_jobs" | jq -er --arg name "$expected_job_name" --argjson attempt "$GITHUB_RUN_ATTEMPT" --argjson run "$GITHUB_RUN_ID" '
+              [.[] | select(.name == $name and .run_attempt == $attempt and .run_id == $run)]
+              | select(length == 1) | .[0].id | select(type == "number" and . > 0 and . == floor)
+            ')"; then
+              echo "::error::CodeQL settlement rejected missing exact handler job ID."
+              exit 1
+            fi
+            if ! [[ "$handler_job_id" =~ ^[1-9][0-9]*$ ]]; then
+              echo "::error::CodeQL settlement rejected missing exact handler job ID."
+              exit 1
+            fi
+            receipt_log="$(mktemp "$RUNNER_TEMP/codeql-sarif-receipt.XXXXXX")"
+            if ! GH_TOKEN="$HANDLER_READ_TOKEN" gh api --allow-escape-sequences \
+              "repos/${GITHUB_REPOSITORY}/actions/jobs/${handler_job_id}/logs" >"$receipt_log"; then
+              echo "::error::CodeQL settlement could not read authenticated SARIF receipt."
+              exit 1
+            fi
+            GH_TOKEN="$SARIF_READ_TOKEN" REPOSITORY="$TARGET_REPOSITORY" LANGUAGE="$language" \
+              RUN_ID="$GITHUB_RUN_ID" RUN_ATTEMPT="$GITHUB_RUN_ATTEMPT" \
+              python3 -I "$RUNNER_TEMP/codeql_sarif_publication.py" verify-log --identity-env --log "$receipt_log"
             clean_gate_count="$(printf '%s' "$handler_jobs" | jq --arg name "$expected_job_name" --argjson attempt "$GITHUB_RUN_ATTEMPT" '
               [.[] | select(
                 .name == $name
diff --git a/docs/doctoring/codeql-direct-sarif-publication.md b/docs/doctoring/codeql-direct-sarif-publication.md
new file mode 100644
index 0000000000..66e00da5cb
--- /dev/null
+++ b/docs/doctoring/codeql-direct-sarif-publication.md
@@ -0,0 +1,83 @@
+# CodeQL dispatch direct SARIF publication
+
+The central `codeql-scan-dispatch.yml` successor replaces its failed Actions
+`CreateArtifact` storage dependency with GitHub's documented code-scanning SARIF
+REST API. It does not declare artifact storage repaired. #2607 supplies precedent
+for direct code-scanning, not proof that this central consumer has run.
+
+## Acceptance order
+
+1. Existing trusted default-branch dispatch and live target metadata admission.
+2. Existing pinned CodeQL analysis with `upload: false` and unchanged Medium+
+   gate. Only trusted pinned helpers execute with Python isolated imports.
+3. One public REST `POST /repos/{target}/code-scanning/sarifs`, using `commit_sha`
+   (not the action-internal `commit_oid`), `refs/pull/{number}/head`, gzip/base64,
+   `tool_name: CodeQL`, local checkout URI and `validate: true`. Preserve category
+   `/language:{language}` via SARIF `automationDetails.id = /language:{language}/`.
+   Never impersonate the default-setup analysis key; API readback records the
+   actual key. An existing different category fails before POST.
+4. Poll the returned upload ID with bounded GETs. Require `complete`, no errors,
+   exactly one associated analysis and a second exact analysis GET. Verify
+   upload ID, head commit, PR ref, CodeQL tool/category, nonempty actual analysis
+   key and no analysis error **or warning** (warnings may mean data truncation).
+   The processed SARIF GET is a subset, not the original raw evidence.
+5. Existing GHAS base/head configuration-identity continuity, unchanged. A new
+   API category/key cannot stand in for another protected-base identity. Default
+   setup was `not-configured` in the retained target receipt; never disable or
+   change it here. Different target setup/identity remains a real admission gate.
+6. Rehash custody and recheck the actual processed analysis in the established
+   `Preserve CodeQL SARIF evidence` step before existing status publication.
+7. Exact-run settlement reads the authenticated scan job's log receipt and
+   independently rechecks that upload/analysis with a separate read credential.
+   The required receiver keeps its established Medium/GHAS/preservation step
+   names. No synthetic status, altered required-check definition or gate bypass.
+
+## Evidence and retention
+
+Each invocation creates an exclusive, mode0700 directory outside checkout and
+`RUNNER_TEMP`:
+`$(dirname "$RUNNER_WORKSPACE")/.cwl-codeql-sarif/{run}/{attempt}/{language}`.
+It fsyncs exact original SARIF, normalized upload SARIF, processed SARIF subset,
+analysis JSON and a stateful receipt. File modes are0600. Repeated custody roots
+are refused before POST; there is no automatic POST replay, token fallback,
+cleanup, deletion or retention sweep. Failure after request intent leaves
+`incomplete-no-replay`, raw bytes and any returned upload ID for the custodian.
+
+The bounded `CWL_CODEQL_SARIF_RECEIPT` job-log line and job summary contain only
+identity, hashes and accepted upload/analysis references, not findings/tokens.
+The log is authenticated by exact run/attempt/job selection. Settlement does not
+trust that text alone: it GETs the upload/analysis again. A missing log, failed
+processing, mismatch or lost receipt blocks settlement. GitHub logs follow the
+repository's retention (observed90 days); self-hosted original-byte custody has
+**minimum7-day retention, no deletion in this implementation**. The lifecycle
+parent must admit this path to the existing durable host custody/backup policy
+before operational use, particularly for ephemeral workers. Local fsync is not
+an off-host backup or a guarantee that an operator will not delete the host.
+
+## Permission boundary and rollout
+
+Same-repository scans request job `security-events: write`; no credentials are
+changed. Cross-repository scans request one separate token from the existing
+owned Noema App, limited to one target and only `security-events: write`. Existing
+analysis-read, status-write and Actions-write tokens remain separate. An
+installation lacking the new write grant fails closed; its owner must explicitly
+approve that capability and accept the installation update. This source does
+not change App settings, copy a personal token or infer write authority from a
+successful analyses read. ADR-0033's earlier no-code-scanning-write decision is
+superseded **only for this separately reviewed direct-SARIF successor**, not as
+an assertion that the grant already exists.
+
+Publish normally only after independent exact-source review and current writer
+reconciliation (notably #2657's live-base metadata fix). Then protected adoption,
+owned permission/custody admission and one separately authorized fresh exact-head
+run are distinct gates. This private implementation authorizes none of those
+remote operations; historical four quota-failed jobs stay failed. Shared
+self-hosted runner admission, Medium threshold and existing lifecycle parent are
+unchanged. No new runner, scheduler or framework is introduced.
+
+## Official contract
+
+- https://docs.github.com/en/rest/code-scanning/code-scanning#upload-an-analysis-as-sarif-data
+- https://docs.github.com/en/rest/code-scanning/code-scanning#get-information-about-a-sarif-upload
+- https://docs.github.com/en/rest/code-scanning/code-scanning#get-a-code-scanning-analysis-for-a-repository
+- https://docs.github.com/en/code-security/code-scanning/integrating-with-code-scanning/sarif-support-for-code-scanning
diff --git a/scripts/ci/codeql_sarif_publication.py b/scripts/ci/codeql_sarif_publication.py
new file mode 100644
index 0000000000..6d283160cc
--- /dev/null
+++ b/scripts/ci/codeql_sarif_publication.py
@@ -0,0 +1,372 @@
+#!/usr/bin/env python3
+"""Publish exact-head CodeQL SARIF via supported REST, retaining fail-closed custody.
+
+This helper never posts statuses or wakes runs. Only the caller's Medium+ gate
+and unchanged GHAS base/head identity proof can authorize those transitions.
+A 202 upload is not completion. An authenticated job-log receipt is usable only
+after rechecking the upload and analysis. Original bytes stay in an exclusive,
+non-temporary self-hosted custody directory; no deletion/retention sweep exists.
+"""
+from __future__ import annotations
+
+import argparse
+import base64
+import gzip
+import hashlib
+import json
+import math
+import os
+from pathlib import Path
+import re
+import sys
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+PREFIX = "CWL_CODEQL_SARIF_RECEIPT "
+SCHEMA = "cwl-codeql-sarif-v1"
+MAX_BYTES = 64 * 1024 * 1024
+MAX_RESPONSE = 8 * 1024 * 1024
+LANGUAGES = {"actions", "c-cpp", "csharp", "go", "java-kotlin", "javascript-typescript", "python", "ruby", "rust", "swift"}
+
+
+class PublicationError(RuntimeError):
+    """Fixed, values-free error; never echo API diagnostics or bearer data."""
+
+
+class RejectRedirects(urllib.request.HTTPRedirectHandler):
+    """Do not forward bearer headers on redirects."""
+    def redirect_request(self, _request, _fp, _code, _message, _headers, _url):
+        """Refuse every redirect."""
+        return None
+
+
+OPENER = urllib.request.build_opener(RejectRedirects())
+
+
+def load_json(raw):
+    """Reject malformed, duplicate-key and nonfinite evidence."""
+    def pairs(items):
+        """Preserve unique object membership."""
+        out = {}
+        for key, value in items:
+            if key in out:
+                raise PublicationError("duplicate evidence key")
+            out[key] = value
+        return out
+    def constant(_value):
+        """Reject nonfinite JSON numbers."""
+        raise PublicationError("nonfinite evidence")
+    def finite_float(value):
+        """Reject exponent overflow as well as literal nonfinite constants."""
+        number = float(value)
+        if not math.isfinite(number):
+            raise PublicationError("nonfinite evidence")
+        return number
+    try:
+        return json.loads(raw, object_pairs_hook=pairs, parse_constant=constant, parse_float=finite_float)
+    except (ValueError, UnicodeError) as exc:
+        raise PublicationError("malformed evidence JSON") from exc
+
+
+class GitHubAPI:
+    """Single-attempt, fixed-origin REST actor; no mutation retry or token fallback."""
+    def __init__(self, token):
+        """Require a caller-provided target-scoped permission."""
+        self.token = token
+
+    def __call__(self, method, path, *, payload=None, accept="application/vnd.github+json"):
+        """Use the documented public SARIF/analyses API only."""
+        if not self.token:
+            raise PublicationError("target security-events token missing")
+        if (not isinstance(path, str) or re.fullmatch(
+            r"repos/[A-Za-z0-9_-]+/[A-Za-z0-9_.-]+/code-scanning/(?:sarifs(?:/[A-Za-z0-9_-]+)?|analyses(?:/[1-9][0-9]*|\?sarif_id=[A-Za-z0-9_-]+&per_page=100))", path
+        ) is None or method not in {"GET", "POST"}):
+            raise PublicationError("untrusted API endpoint")
+        request = urllib.request.Request("https://api.github.com/" + path,
+            data=json.dumps(payload, allow_nan=False).encode() if payload is not None else None,
+            headers={"Accept": accept, "Authorization": "Bearer " + self.token,
+                     "Content-Type": "application/json", "X-GitHub-Api-Version": "2022-11-28",
+                     "User-Agent": "cwl-codeql-sarif-publication"}, method=method)
+        try:
+            with OPENER.open(request, timeout=30) as response:
+                expected = 202 if method == "POST" else 200
+                if response.status != expected or response.headers.get("Link"):
+                    raise PublicationError("unexpected API status or incomplete evidence")
+                raw = response.read(MAX_RESPONSE + 1)
+                if len(raw) > MAX_RESPONSE:
+                    raise PublicationError("API evidence exceeds bound")
+        except urllib.error.HTTPError as exc:
+            exc.close()
+            raise PublicationError("GitHub SARIF API HTTP refusal") from None
+        except (urllib.error.URLError, TimeoutError, OSError):
+            raise PublicationError("GitHub SARIF API transport failed; no replay") from None
+        return load_json(raw)
+
+
+def validate_identity(identity):
+    """Validate the admitted source/run/PR tuple before creating any request."""
+    keys = {"repository", "pr_number", "base_sha", "head_sha", "language", "workflow_sha", "run_id", "run_attempt"}
+    if not isinstance(identity, dict) or set(identity) != keys or any(type(x) is not str for x in identity.values()):
+        raise PublicationError("invalid publication identity")
+    if not re.fullmatch(r"ContextualWisdomLab/[A-Za-z0-9_.-]+", identity["repository"]) or identity["repository"].split("/")[1] in {".", ".."}:
+        raise PublicationError("invalid target repository")
+    for field in ("head_sha", "base_sha", "workflow_sha"):
+        if not re.fullmatch(r"[a-f0-9]{40}", identity[field]):
+            raise PublicationError("invalid commit identity")
+    for field in ("pr_number", "run_id", "run_attempt"):
+        if not re.fullmatch(r"[1-9][0-9]{0,19}", identity[field]):
+            raise PublicationError("invalid run identity")
+    if identity["language"] not in LANGUAGES:
+        raise PublicationError("unsupported CodeQL language")
+
+
+def category(identity):
+    """Retain the existing per-language category, not a run-specific replacement."""
+    return "/language:" + identity["language"]
+
+
+def digest(raw):
+    """Hash exact retained bytes."""
+    return hashlib.sha256(raw).hexdigest()
+
+
+def sync_directory(root):
+    """Commit directory entries, propagating any durability failure."""
+    dirfd = os.open(root, os.O_RDONLY)
+    try:
+        os.fsync(dirfd)
+    finally:
+        os.close(dirfd)
+
+
+def save(root, name, raw):
+    """Durably write an owned evidence file without following a leaf symlink."""
+    path = root / name
+    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW
+    fd = os.open(path, flags, 0o600)
+    with os.fdopen(fd, "wb") as stream:
+        stream.write(raw)
+        stream.flush()
+        os.fsync(stream.fileno())
+    sync_directory(root)
+
+
+def save_receipt(root, receipt):
+    """Write current custody state before and after the single POST."""
+    save(root, "receipt.json", json.dumps(receipt, sort_keys=True, allow_nan=False).encode() + b"\n")
+
+
+def prepare(raw, identity):
+    """Normalize only absent automationDetails; never overwrite another category."""
+    data = load_json(raw)
+    if not isinstance(data, dict) or data.get("version") != "2.1.0" or not isinstance(data.get("runs"), list) or len(data["runs"]) != 1:
+        raise PublicationError("one CodeQL SARIF run required per language")
+    run = data["runs"][0]
+    if not isinstance(run, dict) or run.get("tool", {}).get("driver", {}).get("name") != "CodeQL":
+        raise PublicationError("CodeQL SARIF driver required")
+    wanted = category(identity) + "/"
+    details = run.get("automationDetails")
+    if details is None:
+        run["automationDetails"] = {"id": wanted}
+    elif not isinstance(details, dict) or details.get("id") != wanted:
+        raise PublicationError("conflicting CodeQL SARIF category")
+    return json.dumps(data, sort_keys=True, separators=(",", ":"), allow_nan=False).encode()
+
+
+def processing(api, endpoint, *, attempts, sleeper):
+    """Bounded GET polling; partial/failed/unknown states cannot become accepted."""
+    for attempt in range(attempts):
+        data = api("GET", endpoint)
+        if not isinstance(data, dict) or data.get("errors") not in (None, []):
+            raise PublicationError("SARIF processing errors")
+        state = data.get("processing_status")
+        if state == "complete":
+            return
+        if state != "pending":
+            raise PublicationError("SARIF processing failed or unknown")
+        if attempt + 1 < attempts:
+            sleeper(5)
+    raise PublicationError("SARIF processing deadline exhausted")
+
+
+def check_analysis(row, identity, sarif_id):
+    """Require this exact upload, commit, PR ref, language and complete CodeQL data."""
+    if not isinstance(row, dict):
+        raise PublicationError("invalid code-scanning analysis")
+    if (type(row.get("id")) is not int or row["id"] <= 0 or row.get("sarif_id") != sarif_id
+        or row.get("commit_sha") != identity["head_sha"]
+        or row.get("ref") != "refs/pull/" + identity["pr_number"] + "/head"
+        or row.get("category") != category(identity)
+        or not isinstance(row.get("analysis_key"), str) or not row["analysis_key"]
+        or not isinstance(row.get("tool"), dict) or row["tool"].get("name") != "CodeQL"
+        or row.get("error") != "" or row.get("warning") != ""):
+        raise PublicationError("code-scanning analysis identity or completeness mismatch")
+
+
+def publish(source, root, identity, *, api, checkout_uri, sleeper=time.sleep):
+    """Retain original/upload/readback bytes and return only a fully processed receipt."""
+    validate_identity(identity)
+    paths = sorted(source.rglob("*.sarif"))
+    if len(paths) != 1 or paths[0].is_symlink() or not paths[0].is_file() or not 0 < paths[0].stat().st_size <= MAX_BYTES:
+        raise PublicationError("exactly one bounded SARIF file required")
+    raw = paths[0].read_bytes()
+    if not checkout_uri.startswith("file:///"):
+        raise PublicationError("local checkout URI required")
+    # Non-temporary host custody is exclusive to this run/attempt/language. A
+    # repeated invocation fails before POST rather than replaying a lost ACK.
+    for parent in (root, *root.parents):
+        if parent.is_symlink():
+            raise PublicationError("custody symlink refused")
+    # Include the first existing (admitted durable) anchor: syncing a new
+    # directory's contents alone does not persist its name in its parent.
+    namespace_parents = [root.parent]
+    while not namespace_parents[-1].exists():
+        namespace_parents.append(namespace_parents[-1].parent)
+    root.parent.mkdir(parents=True, exist_ok=True)
+    root.mkdir(mode=0o700, exist_ok=False)
+    for parent in namespace_parents:
+        sync_directory(parent)
+    record = {"source_sha256": digest(raw)}
+    receipt = {"schema": SCHEMA, "state": "prepared", "identity": identity,
+               "minimum_retention_days": 7, "uploads": [record]}
+    save(root, "source-0.sarif", raw)
+    save_receipt(root, receipt)
+    endpoint = "repos/" + identity["repository"] + "/code-scanning/"
+    try:
+        prepared = prepare(raw, identity)
+        record["upload_sha256"] = digest(prepared)
+        save(root, "upload-0.sarif", prepared)
+        receipt["state"] = "post-intent"
+        save_receipt(root, receipt)
+        response = api("POST", endpoint + "sarifs", payload={
+            "commit_sha": identity["head_sha"], "ref": "refs/pull/" + identity["pr_number"] + "/head",
+            "sarif": base64.b64encode(gzip.compress(prepared, mtime=0)).decode(),
+            "tool_name": "CodeQL", "checkout_uri": checkout_uri, "validate": True})
+        sarif_id = response.get("id") if isinstance(response, dict) else None
+        if not isinstance(sarif_id, str) or re.fullmatch(r"[A-Za-z0-9_-]{1,100}", sarif_id) is None:
+            raise PublicationError("invalid SARIF upload acknowledgement")
+        record["sarif_id"] = sarif_id
+        receipt["state"] = "accepted-not-processed"
+        save_receipt(root, receipt)
+        processing(api, endpoint + "sarifs/" + sarif_id, attempts=24, sleeper=sleeper)
+        rows = api("GET", endpoint + "analyses?sarif_id=" + sarif_id + "&per_page=100")
+        if not isinstance(rows, list) or len(rows) != 1:
+            raise PublicationError("ambiguous or missing upload analysis")
+        check_analysis(rows[0], identity, sarif_id)
+        row = api("GET", endpoint + "analyses/" + str(rows[0]["id"]))
+        check_analysis(row, identity, sarif_id)
+        if row != rows[0]:
+            raise PublicationError("analysis changed during readback")
+        record["analysis_id"] = row["id"]
+        record["analysis_key"] = row["analysis_key"]
+        save(root, "analysis-0.json", json.dumps(row, sort_keys=True).encode())
+        subset = api("GET", endpoint + "analyses/" + str(row["id"]), accept="application/sarif+json")
+        if not isinstance(subset, dict) or subset.get("version") != "2.1.0" or not subset.get("runs"):
+            raise PublicationError("processed SARIF evidence unavailable")
+        subset_raw = json.dumps(subset, sort_keys=True).encode()
+        save(root, "processed-0.sarif", subset_raw)
+        record["processed_subset_sha256"] = digest(subset_raw)
+        receipt["state"] = "complete"
+        save_receipt(root, receipt)
+        return receipt
+    except Exception:
+        receipt["state"] = "incomplete-no-replay"
+        save_receipt(root, receipt)
+        raise
+
+
+def receipt_line(receipt):
+    """Encode fixed-schema, bounded metadata only, never findings or credentials."""
+    return PREFIX + json.dumps(receipt, sort_keys=True, separators=(",", ":"))
+
+
+def verify_log(log, identity, *, api):
+    """Verify one authenticated job-log receipt against current API data, no writes."""
+    validate_identity(identity)
+    matches = []
+    for line in log.splitlines():
+        # gh job log GET includes an optional RFC3339 timestamp prefix. Anchor
+        # the record, not a shell echo of the command printed by the runner.
+        match = re.fullmatch(r"(?:[0-9T:.Z+-]+\s+)?" + re.escape(PREFIX) + r"(.{1,8192})", line)
+        if match:
+            matches.append(load_json(match.group(1)))
+    if len(matches) != 1:
+        raise PublicationError("missing or ambiguous durable SARIF receipt")
+    receipt = matches[0]
+    if (not isinstance(receipt, dict) or receipt.get("schema") != SCHEMA
+        or receipt.get("state") != "complete" or receipt.get("identity") != identity
+        or receipt.get("minimum_retention_days") != 7
+        or not isinstance(receipt.get("uploads"), list) or len(receipt["uploads"]) != 1):
+        raise PublicationError("durable SARIF receipt identity mismatch")
+    record = receipt["uploads"][0]
+    for field in ("source_sha256", "upload_sha256", "processed_subset_sha256"):
+        if not isinstance(record, dict) or re.fullmatch(r"[a-f0-9]{64}", str(record.get(field, ""))) is None:
+            raise PublicationError("invalid SARIF custody digest")
+    sid = record.get("sarif_id")
+    aid = record.get("analysis_id")
+    if not isinstance(sid, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,100}", sid) or type(aid) is not int or aid <= 0:
+        raise PublicationError("invalid SARIF receipt reference")
+    endpoint = "repos/" + identity["repository"] + "/code-scanning/"
+    processing(api, endpoint + "sarifs/" + sid, attempts=1, sleeper=lambda _: None)
+    row = api("GET", endpoint + "analyses/" + str(aid))
+    check_analysis(row, identity, sid)
+    if row["id"] != aid or row["analysis_key"] != record.get("analysis_key"):
+        raise PublicationError("receipt analysis changed")
+    return receipt
+
+
+def verify_custody(root, identity, *, api):
+    """Rehash all retained bytes before the established evidence step can pass."""
+    receipt = load_json((root / "receipt.json").read_bytes())
+    verify_log(receipt_line(receipt), identity, api=api)
+    record = receipt["uploads"][0]
+    for name, field in (("source-0.sarif", "source_sha256"),
+                        ("upload-0.sarif", "upload_sha256"),
+                        ("processed-0.sarif", "processed_subset_sha256")):
+        path = root / name
+        if path.is_symlink() or digest(path.read_bytes()) != record[field]:
+            raise PublicationError("custody bytes changed")
+    return receipt
+
+
+def main(argv=None):
+    """CLI run only from pinned trusted bytes under Python isolated mode."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("mode", choices=("publish", "verify-log", "verify-custody"))
+    parser.add_argument("--identity-env", action="store_true", required=True)
+    parser.add_argument("--source", type=Path)
+    parser.add_argument("--custody", type=Path)
+    parser.add_argument("--log", type=Path)
+    args = parser.parse_args(argv)
+    identity = {key: os.environ.get(key.upper(), "") for key in (
+        "repository", "pr_number", "base_sha", "head_sha", "language", "workflow_sha", "run_id", "run_attempt")}
+    try:
+        api = GitHubAPI(os.environ.get("GH_TOKEN", ""))
+        if args.mode == "publish":
+            if args.source is None or args.custody is None:
+                raise PublicationError("source and exclusive custody required")
+            receipt = publish(args.source, args.custody, identity, api=api,
+                              checkout_uri=Path.cwd().resolve().as_uri() + "/")
+            print(receipt_line(receipt), flush=True)
+            summary = os.environ.get("GITHUB_STEP_SUMMARY")
+            if summary:
+                with open(summary, "a", encoding="utf-8") as stream:
+                    stream.write("\n### CodeQL target SARIF receipt\n```json\n" + json.dumps(receipt, sort_keys=True) + "\n```\n")
+        elif args.mode == "verify-custody":
+            if args.custody is None:
+                raise PublicationError("custody required")
+            verify_custody(args.custody, identity, api=api)
+        else:
+            if args.log is None or args.log.stat().st_size > MAX_RESPONSE:
+                raise PublicationError("bounded job log required")
+            verify_log(args.log.read_text(), identity, api=api)
+        return 0
+    except (PublicationError, OSError, ValueError, TypeError, AttributeError):
+        print("::error::CodeQL SARIF publication/custody incomplete; no success or replay authorized.", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/test_codeql_sarif_publication.py b/tests/test_codeql_sarif_publication.py
new file mode 100644
index 0000000000..9e7805f048
--- /dev/null
+++ b/tests/test_codeql_sarif_publication.py
@@ -0,0 +1,505 @@
+"""Offline real HTTP-adapter/CLI controls for direct CodeQL SARIF custody."""
+from __future__ import annotations
+
+import base64
+import copy
+import gzip
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / "scripts/ci/codeql_sarif_publication.py"
+
+
+def module():
+    """Load the exact trusted helper without a network actor."""
+    spec = importlib.util.spec_from_file_location("publication", SCRIPT)
+    assert spec is not None and spec.loader is not None
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+def identity():
+    """Synthetic admitted metadata, never an actual GitHub run."""
+    return dict(repository="ContextualWisdomLab/.github", pr_number="42",
+                base_sha="a" * 40, head_sha="b" * 40, language="python",
+                workflow_sha="c" * 40, run_id="100", run_attempt="1")
+
+
+class Actor:
+    """Finite supported REST responses; every unexpected request fails."""
+    def __init__(self, status=None, analysis=None):
+        self.calls = []
+        self.status = status or {"processing_status": "complete"}
+        self.analysis = analysis or dict(id=17, sarif_id="upload-1", ref="refs/pull/42/head",
+            commit_sha="b" * 40, category="/language:python", analysis_key="API",
+            tool={"name": "CodeQL", "version": "2.0"}, error="", warning="")
+
+    def __call__(self, method, path, *, payload=None, accept="application/vnd.github+json"):
+        self.calls.append((method, path, payload, accept))
+        if method == "POST":
+            assert path == "repos/ContextualWisdomLab/.github/code-scanning/sarifs"
+            assert set(payload) == {"commit_sha", "ref", "sarif", "tool_name", "checkout_uri", "validate"}
+            assert payload["commit_sha"] == "b" * 40
+            assert payload["ref"] == "refs/pull/42/head"
+            data = json.loads(gzip.decompress(base64.b64decode(payload["sarif"])))
+            assert data["runs"][0]["automationDetails"]["id"] == "/language:python/"
+            return {"id": "upload-1", "url": "https://api.github.com/repos/ContextualWisdomLab/.github/code-scanning/sarifs/upload-1"}
+        if path.endswith("/sarifs/upload-1"):
+            return copy.deepcopy(self.status)
+        if "analyses?sarif_id=upload-1" in path:
+            return [copy.deepcopy(self.analysis)]
+        if path.endswith("/analyses/17"):
+            if accept == "application/sarif+json":
+                return {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL"}}, "results": []}]}
+            return copy.deepcopy(self.analysis)
+        raise AssertionError((method, path))
+
+
+def publish(tmp_path, actor=None, content=None, **changes):
+    """Exercise the production publisher with synthetic external HTTP only."""
+    m = module()
+    source = tmp_path / "sarif"
+    source.mkdir()
+    raw = json.dumps(content or {"version": "2.1.0", "runs": [{
+        "tool": {"driver": {"name": "CodeQL", "rules": []}}, "results": []}]}).encode()
+    (source / "python.sarif").write_bytes(raw)
+    args = identity() | changes
+    receipt = m.publish(source, tmp_path / "custody", args, api=actor or Actor(),
+                        checkout_uri=tmp_path.as_uri() + "/", sleeper=lambda _: None)
+    return receipt, raw
+
+
+def test_supported_post_processing_identity_and_durable_bytes(tmp_path):
+    actor = Actor()
+    receipt, raw = publish(tmp_path, actor)
+    assert receipt["schema"] == "cwl-codeql-sarif-v1"
+    assert receipt["state"] == "complete"
+    assert receipt["identity"] == identity()
+    record = receipt["uploads"][0]
+    assert record["source_sha256"] == hashlib.sha256(raw).hexdigest()
+    assert record["sarif_id"] == "upload-1"
+    assert record["analysis_id"] == 17
+    assert (tmp_path / "custody" / "source-0.sarif").read_bytes() == raw
+    assert json.loads((tmp_path / "custody" / "receipt.json").read_text()) == receipt
+    assert sum(c[0] == "POST" for c in actor.calls) == 1
+
+
+@pytest.mark.parametrize("status", [
+    {"processing_status": "failed", "errors": ["secret diagnostic"]},
+    {"processing_status": "complete", "errors": ["partial processing"]},
+    {"processing_status": "pending"}, {"processing_status": "mystery"}, {},
+])
+def test_processing_unknown_failure_partial_and_deadline_fail_closed(tmp_path, status):
+    actor = Actor(status=status)
+    actor.status = status
+    with pytest.raises(RuntimeError):
+        publish(tmp_path, actor)
+    receipt = json.loads((tmp_path / "custody" / "receipt.json").read_text())
+    assert receipt["state"] != "complete"
+    assert "secret diagnostic" not in json.dumps(receipt)
+    assert sum(c[0] == "POST" for c in actor.calls) == 1
+
+
+@pytest.mark.parametrize("field,value", [
+    ("commit_sha", "d" * 40), ("ref", "refs/heads/feature"),
+    ("category", "/language:actions"), ("sarif_id", "another-upload"),
+    ("error", "failed"), ("warning", "truncated"), ("id", True),
+    ("tool", {"name": "Other"}), ("analysis_key", ""),
+])
+def test_analysis_readback_exact_identity_and_no_partial_acceptance(tmp_path, field, value):
+    actor = Actor()
+    actor.analysis[field] = value
+    with pytest.raises(RuntimeError):
+        publish(tmp_path, actor)
+    assert json.loads((tmp_path / "custody" / "receipt.json").read_text())["state"] != "complete"
+
+
+@pytest.mark.parametrize("details", [{"id": "/language:actions/"}, {"id": "/language:python/unreviewed"}])
+def test_existing_conflicting_category_not_silently_overwritten(tmp_path, details):
+    actor = Actor()
+    with pytest.raises(RuntimeError):
+        publish(tmp_path, actor, content={"version": "2.1.0", "runs": [{
+            "tool": {"driver": {"name": "CodeQL"}}, "automationDetails": details, "results": []}]})
+    assert not actor.calls
+
+
+def test_http_adapter_canonical_authority_no_redirect_and_no_retry(monkeypatch):
+    m = module()
+    calls = []
+    class Response:
+        """Actual Request boundary observer."""
+        status = 202
+        headers = {}
+        def __enter__(self): return self
+        def __exit__(self, *_): return None
+        def read(self, *_): return b'{"id":"upload-1"}'
+    def open_request(request, **kwargs):
+        calls.append(request)
+        return Response()
+    monkeypatch.setattr(m.OPENER, "open", open_request)
+    api = m.GitHubAPI("synthetic-token")
+    assert api("POST", "repos/ContextualWisdomLab/.github/code-scanning/sarifs", payload={}) == {"id": "upload-1"}
+    assert calls[0].full_url.startswith("https://api.github.com/repos/")
+    assert json.loads(calls[0].data) == {}
+    with pytest.raises(RuntimeError): api("GET", "https://attacker.invalid/token")
+    assert len(calls) == 1
+    assert m.RejectRedirects().redirect_request(None, None, 302, "", {}, "https://attacker.invalid") is None
+
+
+def test_receipt_log_verifier_rechecks_upload_and_exact_analysis(tmp_path):
+    m = module()
+    receipt, _ = publish(tmp_path)
+    log = "2026-10-26T00:00:00Z " + m.receipt_line(receipt) + "\n"
+    result = m.verify_log(log, identity(), api=Actor())
+    assert result == receipt
+    for field, value in [("head_sha", "d" * 40), ("run_id", "101"), ("run_attempt", "2"), ("language", "actions"), ("workflow_sha", "e" * 40)]:
+        with pytest.raises(RuntimeError): m.verify_log(log, identity() | {field: value}, api=Actor())
+    with pytest.raises(RuntimeError): m.verify_log(log + log, identity(), api=Actor())
+    with pytest.raises(RuntimeError): m.verify_log(log, identity(), api=Actor(status={"processing_status": "failed"}))
+
+
+def test_no_post_replay_after_ambiguous_transport(tmp_path):
+    actor = Actor()
+    def fail(method, path, **kwargs):
+        actor.calls.append((method, path))
+        raise RuntimeError("synthetic lost acknowledgement")
+    with pytest.raises(RuntimeError): publish(tmp_path, fail)
+    assert len(actor.calls) == 1
+    assert (tmp_path / "custody" / "source-0.sarif").is_file()
+    assert (tmp_path / "custody" / "receipt.json").is_file()
+
+
+def test_workflow_wires_real_publication_before_identity_without_artifact_quota():
+    text = (ROOT / ".github/workflows/codeql-scan-dispatch.yml").read_text()
+    scan = text.split("  scan:\n", 1)[1].split("  settle-required-run:", 1)[0]
+    assert "security-events: write" in scan
+    assert "permission-security-events: write" in scan
+    assert "permission-security-events: read" in scan
+    assert "Publish and verify CodeQL SARIF in target code scanning" in scan
+    assert scan.index("Publish and verify CodeQL SARIF in target code scanning") < scan.index("Verify GHAS base/head CodeQL configuration identity")
+    assert "actions/upload-artifact@" not in scan
+    assert "--identity-env" in scan
+    assert "python3 -I" in scan
+    settle = text.split("  settle-required-run:", 1)[1]
+    assert "/artifacts?" not in settle
+    assert "verify-log" in settle
+    assert "codeql_sarif_publication.py" in settle
+
+
+@pytest.mark.parametrize("name", ["source-0.sarif", "upload-0.sarif", "processed-0.sarif"])
+def test_custody_rehash_rejects_changed_original_upload_or_subset(tmp_path, name):
+    receipt, _ = publish(tmp_path)
+    m = module()
+    assert m.verify_custody(tmp_path / "custody", identity(), api=Actor()) == receipt
+    (tmp_path / "custody" / name).write_bytes(b"changed")
+    with pytest.raises(RuntimeError):
+        m.verify_custody(tmp_path / "custody", identity(), api=Actor())
+
+
+def test_existing_custody_blocks_post_replay(tmp_path):
+    publish(tmp_path)
+    actor = Actor()
+    with pytest.raises(FileExistsError):
+        module().publish(tmp_path / "sarif", tmp_path / "custody", identity(), api=actor,
+                         checkout_uri=tmp_path.as_uri() + "/")
+    assert not actor.calls
+
+
+@pytest.mark.parametrize("raw", ['{"x":1,"x":2}', '{"x":NaN}', '{"x":1e999}', 'not-json'])
+def test_malformed_duplicate_and_nonfinite_evidence_rejected(raw):
+    with pytest.raises(RuntimeError): module().load_json(raw)
+
+
+def test_missing_permission_retains_original_without_network(tmp_path, monkeypatch):
+    m = module()
+    monkeypatch.setattr(m.OPENER, "open", lambda *_args, **_kw: pytest.fail("network forbidden"))
+    with pytest.raises(RuntimeError): publish(tmp_path, m.GitHubAPI(""))
+    assert (tmp_path / "custody" / "source-0.sarif").is_file()
+    assert json.loads((tmp_path / "custody" / "receipt.json").read_text())["state"] == "incomplete-no-replay"
+
+
+def test_real_cli_custody_and_log_paths(tmp_path, monkeypatch, capsys):
+    m = module()
+    receipt, _ = publish(tmp_path)
+    monkeypatch.setattr(m, "GitHubAPI", lambda _token: Actor())
+    for key, value in identity().items(): monkeypatch.setenv(key.upper(), value)
+    monkeypatch.setenv("GH_TOKEN", "synthetic-token")
+    assert m.main(["verify-custody", "--identity-env", "--custody", str(tmp_path / "custody")]) == 0
+    log = tmp_path / "job.log"
+    log.write_text(m.receipt_line(receipt))
+    assert m.main(["verify-log", "--identity-env", "--log", str(log)]) == 0
+    monkeypatch.setenv("HEAD_SHA", "d" * 40)
+    assert m.main(["verify-log", "--identity-env", "--log", str(log)]) == 1
+    assert "synthetic-token" not in capsys.readouterr().err
+
+
+@pytest.mark.parametrize("change", [
+    {"extra": "x"}, {"run_id": 1}, {"repository": "Other/repo"},
+    {"repository": "ContextualWisdomLab/.."}, {"head_sha": "invalid"},
+    {"run_attempt": "0"}, {"language": "unknown"},
+])
+def test_identity_rejects_malformed_before_io(change):
+    with pytest.raises(RuntimeError): module().validate_identity(identity() | change)
+
+
+def test_finite_json_and_preexisting_exact_category_remain_supported():
+    m = module()
+    assert m.load_json('{"finite":1.5}') == {"finite": 1.5}
+    data = {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "CodeQL"}},
+            "automationDetails": {"id": "/language:python/"}}]}
+    assert json.loads(m.prepare(json.dumps(data), identity())) == data
+
+
+@pytest.mark.parametrize("data", [[], {"version": "2.1.0", "runs": []},
+    {"version": "2.1.0", "runs": [None]},
+    {"version": "2.1.0", "runs": [{"tool": {"driver": {"name": "Other"}}}]}])
+def test_invalid_sarif_structure_is_not_uploaded(data):
+    with pytest.raises(RuntimeError): module().prepare(json.dumps(data), identity())
+
+
+def test_malformed_analysis_row_rejected():
+    with pytest.raises(RuntimeError): module().check_analysis(None, identity(), "upload-1")
+
+
+def test_source_uri_and_custody_symlink_refusals(tmp_path):
+    m = module()
+    with pytest.raises(RuntimeError):
+        m.publish(tmp_path, tmp_path / "custody", identity(), api=Actor(), checkout_uri=tmp_path.as_uri())
+    source = tmp_path / "sarif"
+    source.mkdir()
+    (source / "test.sarif").write_text('{}')
+    with pytest.raises(RuntimeError):
+        m.publish(source, tmp_path / "custody", identity(), api=Actor(), checkout_uri="https://attacker.invalid")
+    target = tmp_path / "actual"
+    target.mkdir()
+    link = tmp_path / "link"
+    link.symlink_to(target, target_is_directory=True)
+    with pytest.raises(RuntimeError):
+        m.publish(source, link / "custody", identity(), api=Actor(), checkout_uri=tmp_path.as_uri())
+
+
+@pytest.mark.parametrize("fault", ["ack", "rows", "changed", "subset"])
+def test_ambiguous_or_incomplete_readbacks_retain_original(tmp_path, fault):
+    m = module()
+    actor = Actor()
+    def api(method, path, **kwargs):
+        data = actor(method, path, **kwargs)
+        if fault == "ack" and method == "POST": return {"id": "../bad"}
+        if fault == "rows" and "?sarif_id=" in path: return []
+        if fault == "changed" and path.endswith("/analyses/17"): return data | {"new_field": True}
+        if fault == "subset" and kwargs.get("accept") == "application/sarif+json": return {}
+        return data
+    with pytest.raises(RuntimeError): publish(tmp_path, api)
+    assert (tmp_path / "custody" / "source-0.sarif").is_file()
+
+
+@pytest.mark.parametrize("field,value", [("source_sha256", "not-a-hash"),
+    ("sarif_id", "../bad"), ("analysis_id", False), ("analysis_key", "different")])
+def test_receipt_references_and_digests_cannot_be_forged(tmp_path, field, value):
+    receipt, _ = publish(tmp_path)
+    receipt["uploads"][0][field] = value
+    with pytest.raises(RuntimeError): module().verify_log(module().receipt_line(receipt), identity(), api=Actor())
+
+
+def test_nonreceipt_log_lines_are_not_claims(tmp_path):
+    receipt, _ = publish(tmp_path)
+    m = module()
+    assert m.verify_log("ordinary log\n" + m.receipt_line(receipt), identity(), api=Actor()) == receipt
+
+
+@pytest.mark.parametrize("failure", ["status", "link", "size", "http", "transport"])
+def test_http_refusal_bounds_close_and_sanitized_diagnostics(monkeypatch, failure):
+    import io
+    import urllib.error
+    m = module()
+    closed = []
+    class Body(io.BytesIO):
+        def close(self):
+            closed.append(True)
+            super().close()
+    class Response:
+        status = 200
+        headers = {}
+        def __enter__(self): return self
+        def __exit__(self, *_): return None
+        def read(self, *_): return b'{}'
+    response = Response()
+    if failure == "status": response.status = 201
+    if failure == "link": response.headers = {"Link": "untrusted next"}
+    if failure == "size": monkeypatch.setattr(m, "MAX_RESPONSE", 1)
+    def open_request(*_args, **_kwargs):
+        if failure == "http": raise urllib.error.HTTPError("https://api.github.com", 403, "secret", {}, Body(b"secret"))
+        if failure == "transport": raise urllib.error.URLError("secret")
+        return response
+    monkeypatch.setattr(m.OPENER, "open", open_request)
+    with pytest.raises(RuntimeError) as exc:
+        m.GitHubAPI("synthetic-token")("GET", "repos/ContextualWisdomLab/.github/code-scanning/analyses/17")
+    assert "secret" not in str(exc.value)
+    if failure == "http": assert closed
+
+
+def test_cli_publish_summary_and_missing_arguments(tmp_path, monkeypatch, capsys):
+    m = module()
+    monkeypatch.setattr(m, "GitHubAPI", lambda _token: Actor())
+    for key, value in identity().items(): monkeypatch.setenv(key.upper(), value)
+    source = tmp_path / "sarif"
+    source.mkdir()
+    (source / "python.sarif").write_text(json.dumps({"version": "2.1.0", "runs": [
+        {"tool": {"driver": {"name": "CodeQL"}}, "results": []}]}))
+    summary = tmp_path / "summary"
+    monkeypatch.setenv("GITHUB_STEP_SUMMARY", str(summary))
+    assert m.main(["publish", "--identity-env", "--source", str(source), "--custody", str(tmp_path / "custody")]) == 0
+    assert m.PREFIX in capsys.readouterr().out
+    assert "upload-1" in summary.read_text()
+    monkeypatch.delenv("GITHUB_STEP_SUMMARY")
+    assert m.main(["publish", "--identity-env", "--source", str(source), "--custody", str(tmp_path / "second")]) == 0
+    for mode in ("publish", "verify-custody", "verify-log"):
+        assert m.main([mode, "--identity-env"]) == 1
+
+
+def test_real_isolated_entrypoint_missing_permission_exits_one(tmp_path):
+    import os
+    import subprocess
+    import sys
+    process = subprocess.run([sys.executable, "-I", str(SCRIPT), "verify-log", "--identity-env"],
+        env={"PATH": os.environ["PATH"], "TMPDIR": str(tmp_path)}, capture_output=True, text=True)
+    assert process.returncode == 1
+    assert "publication/custody incomplete" in process.stderr
+
+
+def test_native_main_guard_is_executed_in_process(monkeypatch):
+    import runpy
+    import sys
+    monkeypatch.setattr(sys, "argv", [str(SCRIPT), "verify-log", "--identity-env"])
+    monkeypatch.delenv("GH_TOKEN", raising=False)
+    with pytest.raises(SystemExit) as result:
+        runpy.run_path(str(SCRIPT), run_name="__main__")
+    assert result.value.code == 1
+
+
+def observe_publication_fsync(m, monkeypatch, root, failure=None):
+    """Observe real descriptor fsyncs; inject only an explicit IO error, no crash."""
+    import os
+    import stat
+    real_open, real_fsync, real_close = os.open, os.fsync, os.close
+    names, directory_fds, successes, faults = {}, set(), [], []
+    phase = None
+
+    def open_file(path, flags, *args, **kwargs):
+        fd = real_open(path, flags, *args, **kwargs)
+        names[fd] = Path(path).absolute()
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            directory_fds.add(fd)
+        return fd
+
+    def sync(fd):
+        nonlocal phase
+        path = names[fd]
+        if stat.S_ISDIR(os.fstat(fd).st_mode):
+            event = ("directory", path, phase if path == root else "namespace")
+        else:
+            phase = (json.loads(path.read_bytes())["state"] if path.name == "receipt.json"
+                     else path.name)
+            event = ("file", path, phase)
+        if event == failure:
+            faults.append(event)
+            raise OSError("synthetic-fsync-private-diagnostic")
+        real_fsync(fd)
+        successes.append(event)
+
+    def close(fd):
+        real_close(fd)
+        directory_fds.discard(fd)
+
+    monkeypatch.setattr(m.os, "open", open_file)
+    monkeypatch.setattr(m.os, "fsync", sync)
+    monkeypatch.setattr(m.os, "close", close)
+    return successes, faults, directory_fds
+
+
+def namespace_case(tmp_path, existing_depth):
+    """Create only the fixture's admitted anchor; remaining names belong to publish."""
+    source = tmp_path / "input"
+    source.mkdir()
+    raw = json.dumps({"version": "2.1.0", "runs": [
+        {"tool": {"driver": {"name": "CodeQL", "rules": []}}, "results": []}]}).encode()
+    (source / "python.sarif").write_bytes(raw)
+    ancestors = [tmp_path, tmp_path / "persistent", tmp_path / "persistent/100",
+                 tmp_path / "persistent/100/1"]
+    ancestors[existing_depth].mkdir(parents=True, exist_ok=True)
+    root = ancestors[-1] / "python"
+    return source, root, raw, list(reversed(ancestors[existing_depth:]))
+
+
+@pytest.mark.parametrize("existing_depth", [0, 1, 3], ids=["new", "partial", "existing"])
+def test_custody_namespace_entries_and_original_intent_fsynced_before_post(tmp_path, monkeypatch, existing_depth):
+    m = module()
+    source, root, raw, required = namespace_case(tmp_path, existing_depth)
+    successes, faults, directory_fds = observe_publication_fsync(m, monkeypatch, root)
+    actor, snapshot = Actor(), []
+
+    def api(method, path, **kwargs):
+        if method == "POST":
+            snapshot.extend(successes)
+        return actor(method, path, **kwargs)
+
+    receipt = m.publish(source, root, identity(), api=api, checkout_uri=tmp_path.as_uri() + "/")
+    assert sum(c[0] == "POST" for c in actor.calls) == 1
+    for parent in required:
+        event = ("directory", parent, "namespace")
+        assert event in snapshot, "POST without durable custody namespace entry"
+        assert snapshot.index(event) < snapshot.index(("file", root / "receipt.json", "post-intent"))
+    for name, phase in (("source-0.sarif", "source-0.sarif"), ("receipt.json", "prepared"),
+                        ("upload-0.sarif", "upload-0.sarif"), ("receipt.json", "post-intent")):
+        file_event, leaf_event = ("file", root / name, phase), ("directory", root, phase)
+        assert file_event in snapshot and leaf_event in snapshot
+        assert snapshot.index(file_event) < snapshot.index(leaf_event)
+    assert (root / "source-0.sarif").read_bytes() == raw
+    assert receipt["state"] == "complete"
+    assert not faults and not directory_fds
+
+
+FSYNC_FAULTS = [(depth, "namespace", index) for depth in (0, 3) for index in range(4 - depth)]
+FSYNC_FAULTS += [(depth, kind, phase) for depth in (0, 3) for kind in ("file", "leaf")
+                 for phase in ("source-0.sarif", "prepared", "upload-0.sarif", "post-intent")]
+
+
+@pytest.mark.parametrize("existing_depth,kind,stage", FSYNC_FAULTS)
+def test_prepost_fsync_error_zero_post_fixed_diagnostic_and_no_replay(tmp_path, monkeypatch, capsys,
+                                                                    existing_depth, kind, stage):
+    m = module()
+    source, root, raw, required = namespace_case(tmp_path, existing_depth)
+    if kind == "namespace":
+        failure = ("directory", required[stage], "namespace")
+    elif kind == "leaf":
+        failure = ("directory", root, stage)
+    else:
+        name = "receipt.json" if stage in {"prepared", "post-intent"} else stage
+        failure = ("file", root / name, stage)
+    actor = Actor()
+    monkeypatch.setattr(m, "GitHubAPI", lambda _: actor)
+    for key, value in identity().items():
+        monkeypatch.setenv(key.upper(), value)
+    with monkeypatch.context() as io_patch:
+        _successes, faults, directory_fds = observe_publication_fsync(m, io_patch, root, failure)
+        assert m.main(["publish", "--identity-env", "--source", str(source), "--custody", str(root)]) == 1
+    assert faults == [failure], "fault site must actually be exercised"
+    assert not actor.calls and not directory_fds
+    captured = capsys.readouterr()
+    assert captured.out == ""
+    assert captured.err == "::error::CodeQL SARIF publication/custody incomplete; no success or replay authorized.\n"
+    assert root.is_dir()
+    if (root / "source-0.sarif").exists():
+        assert (root / "source-0.sarif").read_bytes() == raw
+    if (root / "receipt.json").exists():
+        assert json.loads((root / "receipt.json").read_bytes())["state"] != "complete"
+    # Retry only after removing the injected error; retain the real exclusive-root fence.
+    with pytest.raises(FileExistsError):
+        m.publish(source, root, identity(), api=actor, checkout_uri=tmp_path.as_uri() + "/")
+    assert not actor.calls
diff --git a/tests/test_codeql_scan_dispatch_workflow_contract.py b/tests/test_codeql_scan_dispatch_workflow_contract.py
index 74c01d7989..9f53e38ee1 100644
--- a/tests/test_codeql_scan_dispatch_workflow_contract.py
+++ b/tests/test_codeql_scan_dispatch_workflow_contract.py
@@ -1013,7 +1013,7 @@ def test_dispatch_publish_keeps_successful_scan_when_status_write_is_denied() ->
 
     assert "GATE_OUTCOME" in publish
     assert 'if [ "$GATE_OUTCOME" = "success" ]; then' in publish
-    assert "exact completed scan and preserved SARIF artifact remain" in publish
+    assert "exact completed scan, durable custody receipt and processed target SARIF remain" in publish
     assert "continue-on-error:" not in publish
     assert "cancel-in-progress: true" not in publish
 
@@ -1086,10 +1086,11 @@ def _run_settlement_step(
     run: dict | None = None,
     required_jobs: list[dict] | None = None,
     handler_jobs: list[dict] | None = None,
-    handler_artifacts: list[dict] | None = None,
+    handler_receipts: list[str] | None = None,
     extra_env: dict[str, str] | None = None,
 ) -> tuple[subprocess.CompletedProcess[str], Path]:
     """Execute the run-wide settlement block against fixture-backed API responses."""
+    tmp_path.mkdir(parents=True, exist_ok=True)
     bash = shutil.which("bash")
     jq = shutil.which("jq")
     assert bash is not None and jq is not None, "bash and jq are required to run this test"
@@ -1165,18 +1166,47 @@ def _run_settlement_step(
             ],
         },
     ]
-    handler_artifacts = handler_artifacts or [
-        {
-            "name": "codeql-dispatch-python-100-1",
-            "expired": False,
-            "size_in_bytes": 10,
-        },
-        {
-            "name": "codeql-dispatch-actions-100-1",
-            "expired": False,
-            "size_in_bytes": 10,
-        },
-    ]
+    # Synthetic REST is the sole external substitution. Execute the real
+    # verifier CLI/main and all receipt/identity predicates in-process.
+    from tests.test_codeql_sarif_publication import Actor, identity, module
+    publication = module()
+    receipts = {}
+    analyses = {}
+    for index, language in enumerate(("python", "actions")):
+        sid = f"upload-{language}"
+        aid = 17 + index
+        receipt = {
+            "schema": publication.SCHEMA, "state": "complete", "minimum_retention_days": 7,
+            "identity": identity() | {"repository": "ContextualWisdomLab/naruon", "language": language},
+            "uploads": [{"sarif_id": sid, "analysis_id": aid, "analysis_key": "API",
+                         "source_sha256": "1" * 64, "upload_sha256": "2" * 64,
+                         "processed_subset_sha256": "3" * 64}],
+        }
+        if handler_receipts is None or language in handler_receipts:
+            receipts[str(200 + index)] = publication.receipt_line(receipt)
+        analyses[str(aid)] = Actor().analysis | {"id": aid, "sarif_id": sid, "category": f"/language:{language}"}
+    for index, job in enumerate(handler_jobs):
+        job.setdefault("id", 200 + index)
+        job.setdefault("run_id", 100)
+    fake_data = tmp_path / "sarif-readbacks.json"
+    fake_data.write_text(json.dumps(analyses))
+    fake_temp = tmp_path / "runner-temp"
+    fake_temp.mkdir()
+    helper = fake_temp / "codeql_sarif_publication.py"
+    helper.write_text(
+        "import importlib.util,json,os,pathlib,sys\n"
+        f"spec=importlib.util.spec_from_file_location('publication', {str(REPO_ROOT / 'scripts/ci/codeql_sarif_publication.py')!r})\n"
+        "m=importlib.util.module_from_spec(spec); spec.loader.exec_module(m)\n"
+        "def api(method,path,**kwargs):\n"
+        " assert method=='GET', 'no mutation in receipt verification'\n"
+        " if '/sarifs/' in path: return {'processing_status':'complete'}\n"
+        " data=json.loads(pathlib.Path(os.environ['FAKE_SARIF_DATA']).read_text())\n"
+        " return data[path.rsplit('/',1)[-1]]\n"
+        "m.GitHubAPI=lambda token: api\n"
+        "raise SystemExit(m.main())\n"
+    )
+    receipt_data = tmp_path / "receipt-logs.json"
+    receipt_data.write_text(json.dumps(receipts))
     script = _extract_run_block(
         WORKFLOW_PATH.read_text(encoding="utf-8"), "Settle exact CodeQL required run"
     )
@@ -1216,7 +1246,7 @@ def _run_settlement_step(
         '  repos/ContextualWisdomLab/naruon/actions/runs/42/jobs*) printf \'%s\\n\' "$FAKE_REQUIRED_JOB_PAGES" ;;\n'
         '  repos/ContextualWisdomLab/naruon/actions/runs/42) printf \'%s\\n\' "$FAKE_RUN_JSON" ;;\n'
         '  repos/ContextualWisdomLab/.github/actions/runs/100/jobs*) printf \'%s\\n\' "$FAKE_HANDLER_JOB_PAGES" ;;\n'
-        '  repos/ContextualWisdomLab/.github/actions/runs/100/artifacts*) printf \'%s\\n\' "$FAKE_HANDLER_ARTIFACT_PAGES" ;;\n'
+        '  repos/ContextualWisdomLab/.github/actions/jobs/*/logs) "$FAKE_PYTHON" -c \'import json,os,sys; print(json.load(open(os.environ["FAKE_RECEIPT_DATA"])).get(sys.argv[1].split("/")[-2], ""))\' "$endpoint" ;;\n'
         "  *) exit 1 ;;\n"
         "esac\n",
         encoding="utf-8",
@@ -1229,9 +1259,12 @@ def _run_settlement_step(
         "FAKE_RUN_JSON": json.dumps(run),
         "FAKE_REQUIRED_JOB_PAGES": "\n".join(json.dumps({"jobs": [job]}) for job in required_jobs),
         "FAKE_HANDLER_JOB_PAGES": json.dumps({"jobs": handler_jobs}),
-        "FAKE_HANDLER_ARTIFACT_PAGES": json.dumps(
-            {"artifacts": handler_artifacts}
-        ),
+        "RUNNER_TEMP": str(fake_temp),
+        "FAKE_RECEIPT_DATA": str(receipt_data),
+        "FAKE_SARIF_DATA": str(fake_data),
+        "FAKE_PYTHON": sys.executable,
+        "SARIF_READ_TOKEN": "synthetic-read-token",
+        "WORKFLOW_SHA": "c" * 40,
         "FAKE_POST_LOG": str(post_log),
         "FAKE_POST_EXIT": "0",
         "FAKE_DENIED_TOKEN": "",
@@ -1533,20 +1566,11 @@ def test_dispatch_settlement_all_mode_reruns_success_and_failure_jobs(tmp_path:
     ]
 
 
-def test_dispatch_settlement_rejects_missing_handler_artifact(tmp_path: Path) -> None:
-    result, post_log = _run_settlement_step(
-        tmp_path,
-        handler_artifacts=[
-            {
-                "name": "codeql-dispatch-python-100-1",
-                "expired": False,
-                "size_in_bytes": 10,
-            }
-        ],
-    )
+def test_dispatch_settlement_rejects_missing_handler_receipt(tmp_path: Path) -> None:
+    result, post_log = _run_settlement_step(tmp_path, handler_receipts=["python"])
 
     assert result.returncode == 1
-    assert "incomplete handler gate or SARIF evidence for actions" in result.stdout
+    assert "publication/custody incomplete" in result.stderr
     assert not post_log.exists()
 
 
@@ -1634,13 +1658,7 @@ def test_dispatch_settlement_rejects_unproven_matrix_subset(tmp_path: Path) -> N
                 "conclusion": "success",
             }
         ],
-        handler_artifacts=[
-            {
-                "name": "codeql-dispatch-actions-100-1",
-                "expired": False,
-                "size_in_bytes": 10,
-            }
-        ],
+        handler_receipts=["actions"],
     )
 
     assert result.returncode == 1