diff --git a/.github/workflows/fragment-offset-qualification.yml b/.github/workflows/fragment-offset-qualification.yml
new file mode 100644
index 00000000..0e53e5e9
--- /dev/null
+++ b/.github/workflows/fragment-offset-qualification.yml
@@ -0,0 +1,82 @@
+name: Fork fragment offset qualification
+
+on:
+  pull_request:
+    types: [opened, reopened, synchronize, labeled]
+
+permissions:
+  contents: read
+
+jobs:
+  qualify:
+    if: >-
+      github.repository == 'mstattma/c2pa-python' &&
+      (github.event.pull_request.author_association == 'OWNER' ||
+       github.event.pull_request.author_association == 'MEMBER' ||
+       github.event.pull_request.author_association == 'COLLABORATOR' ||
+       contains(github.event.pull_request.labels.*.name, 'safe to test'))
+    name: Fragment offset (${{ matrix.os }})
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 90
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-24.04, windows-2022]
+    env:
+      CARGO_INCREMENTAL: "0"
+      CARGO_TERM_COLOR: never
+      PIP_NO_CACHE_DIR: "1"
+      PYTHONUNBUFFERED: "1"
+    defaults:
+      run:
+        shell: bash
+    steps:
+      - name: Checkout exact PR Python
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+          path: candidate
+          persist-credentials: false
+      - name: Checkout exact baseline Python
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+        with:
+          ref: ${{ github.event.pull_request.base.sha }}
+          path: baseline
+          persist-credentials: false
+      - name: Checkout allowlisted native source
+        uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4
+        with:
+          repository: mstattma/c2pa-rs
+          ref: 20d7c45972976c32ab24d5d4ac33e7be826ce1be
+          path: native
+          persist-credentials: false
+      - name: Python 3.12
+        uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+        with:
+          python-version: "3.12"
+      - name: Test qualification helpers (no native imports)
+        run: |
+          set -euo pipefail
+          mkdir -p evidence
+          python -m unittest discover -s candidate/scripts -p test_fragment_offset_qualification.py -v 2>&1 | tee evidence/helper-tests.log
+      - name: Build exact native candidate and record evidence
+        env:
+          CARGO_HOME: ${{ runner.temp }}/fragment-cargo
+        run: python candidate/scripts/fragment_offset_qualification.py build
+      - name: Install Python-only test dependencies
+        run: |
+          set -euo pipefail
+          python -m pip install -r candidate/requirements.txt pytest toml requests
+          python -m pip install --no-deps -e candidate
+      - name: Stock compatibility, same-binary baseline and PR, focused offsets
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: python candidate/scripts/fragment_offset_qualification.py qualify
+      - name: Retain qualification evidence (not a release)
+        if: always()
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
+        with:
+          name: fragment-offset-${{ matrix.os }}-${{ github.event.pull_request.head.sha }}
+          path: evidence/
+          retention-days: 14
+          if-no-files-found: error
diff --git a/docs/usage.md b/docs/usage.md
index c7754e72..f28d8ff3 100644
--- a/docs/usage.md
+++ b/docs/usage.md
@@ -235,6 +235,21 @@ except Exception as err:
     print(err)
 ```
 
+### Verify one fragment of a single-file fragmented MP4
+
+A single-file fragmented MP4 (for example one served as HLS byte ranges) hashes each fragment together with its absolute byte offset in the file. To verify one fragment on its own, pass the bytes before the first C2PA `merkle` box as the init stream, the fragment from its `merkle` box up to the next one, and the fragment's offset in the original file:
+
+```py
+if c2pa.has_fragment_at_offset():
+    with Context() as ctx:
+        with Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(init_bytes), io.BytesIO(fragment_bytes),
+                fragment_offset, context=ctx) as reader:
+            print(reader.get_validation_state())
+```
+
+A wrong cut or offset is reported as `assertion.bmffHash.mismatch` in the validation results rather than raised. An offset of `0` behaves like `Reader.with_fragment()` for fragments stored as separate files. `has_fragment_at_offset()` is `False` when the native library predates this API; the methods then raise `C2paError.NotSupported`.
+
 ### Add a signed manifest to a stream
 
 > [!WARNING]
diff --git a/scripts/fragment_offset_qualification.py b/scripts/fragment_offset_qualification.py
new file mode 100644
index 00000000..23489da0
--- /dev/null
+++ b/scripts/fragment_offset_qualification.py
@@ -0,0 +1,338 @@
+"""Non-publishing fork qualification. Native build mode is GitHub Actions only."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import tomllib
+import xml.etree.ElementTree as ET
+
+
+NATIVE_REPOSITORY = "mstattma/c2pa-rs"
+NATIVE_SHA = "20d7c45972976c32ab24d5d4ac33e7be826ce1be"
+NATIVE_VERSION = "0.92.0-dev"
+FEATURES = "rust_native_crypto,add_thumbnails,http,file_io"
+RUST = "1.96.0"
+SUITES = ("test_unit_tests.py", "test_unit_tests_threaded.py")
+# Reviewed raw JUnit keys: future passing tests may extend, not replace, these IDs.
+BINDING_OFFSET_TEST_IDS = frozenset({
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_exported_from_package",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_false_without_symbols",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_probe_result_is_used_and_cached",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetFeatureDetection::test_unsupported_raises_before_touching_the_reader",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_from_rejects_invalid_context",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_aliased_streams_before_allocation",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_format_encoding_before_allocation",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_format_types_before_allocation",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_invalid_streams_before_allocation",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_non_int_offsets",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetValidation::test_rejects_out_of_range_offsets",
+})
+NATIVE_OFFSET_TEST_IDS = frozenset({
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_from_fails_cleanly_without_a_manifest",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_from_fragment_at_offset_zero",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_maximum_offset_never_verifies",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_native_failure_closes_the_reader",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_replacing_the_fragment_refreshes_results",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_wrong_offset_is_a_mismatch_not_an_error",
+    "tests.test_fragment_at_offset.TestFragmentAtOffsetNative::test_zero_offset_matches_with_fragment",
+})
+REQUIRED_OFFSET_TEST_IDS = BINDING_OFFSET_TEST_IDS | NATIVE_OFFSET_TEST_IDS
+STOCK_NATIVE_SKIP = ("skipped", "pytest.skip", "native library lacks c2pa_reader_with_fragment_at_offset")
+FORK_TEST_ID = "tests.test_unit_tests_threaded.TestForkedChildDoesNotDeadlock::test_parent_copy_unaffected"
+FORK_BASELINE_FAILURE = (
+    "failure", "builtins.AssertionError",
+    'AssertionError: 1 != 0 : Traceback (most recent call last):\n'
+    '  File "<string>", line 11, in <module>\n'
+    "AttributeError: module 'os' has no attribute 'fork'",
+)
+ROOT = Path(__file__).resolve().parents[2]
+EVIDENCE = ROOT / "evidence"
+
+
+def digest(path):
+    return hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+def record(name, value):
+    (EVIDENCE / name).write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
+
+
+def command(args, cwd, name, env=None, test=False):
+    """Only pytest's ordinary failure status is deferred for comparison."""
+    with (EVIDENCE / name).open("w", encoding="utf-8") as log:
+        result = subprocess.run(args, cwd=cwd, env=env, stdout=log,
+                                stderr=subprocess.STDOUT, text=True)
+    print(f"{name}: exit {result.returncode}", flush=True)
+    if result.returncode not in ((0, 1) if test else (0,)):
+        raise subprocess.CalledProcessError(result.returncode, args)
+    return result.returncode
+
+
+def git(path, *args):
+    return subprocess.check_output(["git", "-C", str(path), *args], text=True).strip()
+
+
+def validate_native(path):
+    if os.environ.get("GITHUB_ACTIONS") != "true" or os.environ.get(
+            "GITHUB_REPOSITORY") != "mstattma/c2pa-python":
+        raise RuntimeError("Native qualification builds are fork GitHub Actions only")
+    if not re.fullmatch(r"[0-9a-f]{40}", NATIVE_SHA):
+        raise ValueError("Native SHA must be full lowercase hex")
+    if git(path, "remote", "get-url", "origin") != f"https://github.com/{NATIVE_REPOSITORY}":
+        raise ValueError("Native repository is not allowlisted")
+    if git(path, "rev-parse", "HEAD") != NATIVE_SHA:
+        raise ValueError("Native SHA is not allowlisted")
+    if git(path, "status", "--porcelain"):
+        raise ValueError("Native checkout must be clean before build")
+    if git(path, "ls-files", "Cargo.lock") or (path / "Cargo.lock").exists():
+        raise ValueError("Expected an untracked, absent native Cargo.lock")
+    with (path / "Cargo.toml").open("rb") as file:
+        if tomllib.load(file)["workspace"]["package"]["version"] != NATIVE_VERSION:
+            raise ValueError("Unexpected native version")
+
+
+def build():
+    native = ROOT / "native"
+    validate_native(native)  # Before installing tools or executing any native source.
+    record("sources.json", {
+        "native_repository": NATIVE_REPOSITORY, "native_sha": NATIVE_SHA,
+        "native_version": NATIVE_VERSION, "features": FEATURES,
+        "default_features": False, "profile": "release", "rust": RUST,
+        "python_head": git(ROOT / "candidate", "rev-parse", "HEAD"),
+        "python_base": git(ROOT / "baseline", "rev-parse", "HEAD"),
+        "run_id": os.environ["GITHUB_RUN_ID"], "python": sys.version,
+    })
+    command(["rustup", "toolchain", "install", RUST, "--profile", "minimal"],
+            native, "rustup.log")
+    command(["rustc", f"+{RUST}", "-vV"], native, "rustc.log")
+    cargo = ["cargo", f"+{RUST}"]
+    command(cargo + ["--version", "--verbose"], native, "cargo.log")
+    command(cargo + ["generate-lockfile"], native, "resolve.log")
+    shutil.copy2(native / "Cargo.lock", EVIDENCE / "Cargo.lock")
+    lock_digest = digest(native / "Cargo.lock")
+    record("resolved-lock.json", {"sha256": lock_digest, "generated_in_job": True})
+    flags = ["--locked", "-p", "c2pa-c-ffi", "--no-default-features", "--features", FEATURES]
+    command(cargo + ["tree", *flags, "-e", "features"], native, "features.log")
+    command(cargo + ["build", "--release", *flags], native, "build.log")
+    if digest(native / "Cargo.lock") != lock_digest:
+        raise RuntimeError("Resolved lock changed during locked build")
+    name = "c2pa_c.dll" if sys.platform == "win32" else "libc2pa_c.so"
+    library = EVIDENCE / name
+    shutil.copy2(native / "target" / "release" / name, library)
+    record("native.json", {"library": str(library), "sha256": digest(library),
+                           "resolved_lock_sha256": lock_digest})
+
+
+def suite(source, library, expected_hash, version, test_file, label):
+    # This runs in a fresh process, with PYTHONPATH and the absolute native path set.
+    import c2pa.c2pa as module
+    import pytest
+
+    loaded = Path(module._lib._name)
+    if not loaded.is_absolute() or loaded.resolve() != library:
+        raise RuntimeError(f"Wrong loaded native library: {loaded}")
+    if digest(loaded) != expected_hash or Path(module.__file__).resolve() != source / "src/c2pa/c2pa.py":
+        raise RuntimeError("Native digest or Python module provenance mismatch")
+    if version not in module.sdk_version():
+        raise RuntimeError(f"Unexpected SDK version: {module.sdk_version()}")
+    record(f"{label}-loaded.json", {"module": module.__file__, "library": str(loaded),
+                                   "sha256": digest(loaded), "version": module.sdk_version()})
+
+    class FailureTypes:
+        # Pytest's JUnit omits exception types; preserve them explicitly rather
+        # than considering two different exceptions with the same message equal.
+        @pytest.hookimpl(hookwrapper=True)
+        def pytest_runtest_makereport(self, item, call):
+            report = (yield).get_result()
+            if report.failed and call.excinfo is not None:
+                cls = call.excinfo.type
+                item.user_properties.append(("exception_type", f"{cls.__module__}.{cls.__qualname__}"))
+
+    return pytest.main([f"tests/{test_file}", "-v", "--tb=long",
+                        "-o", "junit_family=xunit2", f"--junitxml={EVIDENCE / (label + '.xml')}"],
+                       plugins=[FailureTypes()])
+
+
+def normalize(message):
+    message = re.sub(r"\\+", "/", message)
+    for folder in (ROOT / "candidate", ROOT / "baseline", ROOT):
+        message = message.replace(folder.as_posix(), "<checkout>")
+    temporary = re.escape(Path(tempfile.gettempdir()).resolve().as_posix())
+    message = re.sub(temporary + r"/tmp[a-zA-Z0-9_-]{8}(?=[/.\s'\"<>:)]|$)", "<temporary>", message)
+    return re.sub(r"0x[0-9a-fA-F]+", "0xADDR", message)
+
+
+def cases(path):
+    result = {}
+    for case in ET.parse(path).iter("testcase"):
+        key = case.attrib["classname"] + "::" + case.attrib["name"]
+        if key in result:
+            raise ValueError(f"Duplicate test ID: {key}")
+        outcome = ("passed", "", "")
+        for kind in ("failure", "error", "skipped"):
+            node = case.find(kind)
+            if node is not None:
+                exception_type = case.find("./properties/property[@name='exception_type']")
+                failure_type = node.get("type", "") if exception_type is None else exception_type.get("value", "")
+                if kind != "skipped" and not failure_type:
+                    raise ValueError(f"Missing exception type for {key}")
+                outcome = (kind, failure_type, normalize(node.get("message", "")))
+                break
+        result[key] = outcome
+    if not result:
+        raise ValueError(f"No tests executed: {path}")
+    return result
+
+
+def focused_problems(report, *, stock=False):
+    problems = [f"Missing required focused test: {key}" for key in REQUIRED_OFFSET_TEST_IDS - report.keys()]
+    for key, outcome in report.items():
+        expected = STOCK_NATIVE_SKIP if stock and key in NATIVE_OFFSET_TEST_IDS else ("passed", "", "")
+        if outcome != expected:
+            problems.append(f"Unexpected focused outcome: {key}: {outcome!r}; expected: {expected!r}")
+    return sorted(problems)
+
+
+def compare(baseline, candidate, *, platform_adjustments=None, context=None):
+    problems = [f"Missing common test: {key}" for key in baseline.keys() - candidate.keys()]
+    for key, outcome in candidate.items():
+        previous = baseline.get(key, ("passed", "", ""))
+        # The base predates the fork guard. Exempt only its exact Windows
+        # missing-fork failure, never a passing test or an unrelated failure.
+        if (platform_adjustments is not None and context in ("stock", "native") and sys.platform == "win32"
+                and not hasattr(os, "fork") and key == FORK_TEST_ID
+                and previous == FORK_BASELINE_FAILURE
+                and outcome == ("skipped", "pytest.skip", "requires os.fork")):
+            platform_adjustments.append({"test_id": key, "platform": sys.platform, "context": context,
+                                         "baseline": previous, "candidate": outcome})
+            continue
+        if outcome[0] != "passed" and outcome != previous:
+            problems.append(f"New/changed outcome: {key}: {outcome!r}; baseline: {previous!r}")
+    return sorted(problems)
+
+
+def qualify():
+    candidate, baseline = ROOT / "candidate", ROOT / "baseline"
+    native = json.loads((EVIDENCE / "native.json").read_text(encoding="utf-8"))
+    library = Path(native["library"]).resolve()
+    # Keep ordinary release pins untouched; only ephemeral preflight files differ.
+    for source in (candidate, baseline):
+        (source / "c2pa-rs-preflight-ref.txt").write_text(f"c2pa-v{NATIVE_VERSION}\n", encoding="utf-8")
+    command([sys.executable, "scripts/download_artifacts.py", "c2pa-v0.91.0"],
+            candidate, "stock-download.log")
+    name = library.name
+    stock = list((candidate / "artifacts").glob(f"*/{name}"))
+    if len(stock) != 1:
+        raise RuntimeError(f"Expected one downloaded stock library, found {stock}")
+    stock = stock[0].resolve()
+    shutil.copy2(stock, EVIDENCE / ("stock-" + name))
+    results, reports, regressions, platform_adjustments = {}, {}, [], []
+    lanes = (("stock-baseline", baseline, stock, digest(stock), "0.91.0"),
+             ("stock", candidate, stock, digest(stock), "0.91.0"),
+             ("baseline", baseline, library, native["sha256"], NATIVE_VERSION),
+             ("candidate", candidate, library, native["sha256"], NATIVE_VERSION))
+    for lane, source, binary, sha, version in lanes:
+        files = SUITES + (("test_fragment_at_offset.py",) if lane in ("stock", "candidate") else ())
+        for test_file in files:
+            label = lane + "-" + test_file.removesuffix(".py")
+            env = dict(os.environ, PYTHONPATH=str(source / "src"),
+                       C2PA_LIBRARY_NAME=str(binary), PYTEST_DISABLE_PLUGIN_AUTOLOAD="1")
+            env.pop("C2PA_PREFLIGHT_RUN", None)
+            env.pop("C2PA_FRAGMENT_OFFSET_REQUIRED", None)
+            if lane in ("baseline", "candidate"):
+                env["C2PA_PREFLIGHT_RUN"] = "1"
+            if lane == "candidate":
+                env["C2PA_FRAGMENT_OFFSET_REQUIRED"] = "1"
+            results[label] = command([sys.executable, str(Path(__file__).resolve()), "suite",
+                                      str(source), str(binary), sha, version, test_file, label],
+                                     source, label + ".log", env, test=True)
+            reports[label] = cases(EVIDENCE / (label + ".xml"))
+            if (results[label] == 0) != all(v[0] in ("passed", "skipped") for v in reports[label].values()):
+                raise RuntimeError(f"Process/JUnit status disagreement: {label}")
+    comparisons = {}
+    for context, base_lane, pr_lane in (("stock", "stock-baseline", "stock"), ("native", "baseline", "candidate")):
+        comparisons[context] = []
+        for file in SUITES:
+            suffix = "-" + file.removesuffix(".py")
+            comparisons[context].extend(compare(reports[base_lane + suffix], reports[pr_lane + suffix],
+                                                platform_adjustments=platform_adjustments, context=context))
+        regressions.extend(f"{context}: {problem}" for problem in comparisons[context])
+    problems = list(regressions)
+    focused = {lane: focused_problems(reports[lane + "-test_fragment_at_offset"], stock=lane == "stock")
+               for lane in ("stock", "candidate")}
+    focused_pass = not focused["candidate"]
+    problems.extend(f"{lane} focused: {problem}" for lane, issues in focused.items() for problem in issues)
+    # The unmodified stock base may fail only the explicitly recognized fork case.
+    allowed_base_failures = {entry["test_id"]: entry["baseline"] for entry in platform_adjustments
+                             if entry["context"] == "stock"}
+    stock_base_problems = []
+    for file in SUITES:
+        label = "stock-baseline-" + file.removesuffix(".py")
+        stock_base_problems.extend(f"Unexpected stock baseline failure: {key}: {outcome!r}"
+                                   for key, outcome in reports[label].items()
+                                   if outcome[0] in ("failure", "error") and allowed_base_failures.get(key) != outcome)
+    problems.extend(stock_base_problems)
+    stock_labels = ["stock-" + file.removesuffix(".py") for file in (*SUITES, "test_fragment_at_offset.py")]
+    stock_pass = (all(results[label] == 0 for label in stock_labels)
+                  and not comparisons["stock"] and not focused["stock"] and not stock_base_problems)
+    if not stock_pass:
+        problems.append("Stock 0.91 compatibility failed")
+    failures = {label: {key: value for key, value in report.items() if value[0] != "passed"}
+                for label, report in reports.items()}
+    record("comparison.json", {"exit_codes": results, "nonpasses": failures,
+                               "regressions": regressions, "qualification_failures": problems,
+                               "comparisons": comparisons, "focused_problems": focused,
+                               "focused_requirements": {"required_ids": sorted(REQUIRED_OFFSET_TEST_IDS),
+                                                        "stock_native_skip_ids": sorted(NATIVE_OFFSET_TEST_IDS),
+                                                        "stock_native_skip": STOCK_NATIVE_SKIP},
+                               "lanes": {lane: {"python_source": str(source), "library": str(binary),
+                                                "sha256": sha, "version": version}
+                                         for lane, source, binary, sha, version in lanes},
+                               "platform_adjustments": platform_adjustments})
+    full = all(status == 0 for label, status in results.items() if label.startswith("candidate-"))
+    summary = (f"## Fragment offset qualification\n\n"
+               f"Candidate full-suite status: **{'PASS' if full else 'FAIL'}**\n\n"
+               f"No-new-regressions: **{'FAIL' if regressions else 'PASS'}**\n\n"
+               f"Stock 0.91 compatibility: **{'PASS' if stock_pass else 'FAIL'}**\n\n"
+               f"Required 18 focused offset IDs (all pass, no skips): **{'PASS' if focused_pass else 'FAIL'}**\n\n"
+               f"Stock focused IDs (11 pass, 7 exact unsupported skips): **{'FAIL' if focused['stock'] else 'PASS'}**\n\n"
+               f"Combined qualification gate: **{'FAIL' if problems else 'PASS'}**\n\n"
+               f"Explicit platform applicability adjustments: **{len(platform_adjustments)}**\n\n"
+               "Full-suite failures are not waived as full passes. Exact IDs, failure types and "
+               "normalized messages (including unchanged baseline failures) are in comparison.json; "
+               "raw JUnit, logs and loaded-library evidence are retained.\n\n"
+               + "\n".join(f"- {label}: exit {status}" for label, status in results.items()) + "\n\n"
+               + "\n".join(f"- `{label} / {key}`: {value!r}"
+                           for label, report in failures.items() for key, value in report.items()
+                           if value[0] != "skipped") + "\n"
+               + "\n".join(f"- Platform applicability ({entry['context']}): `{entry['test_id']}` on {entry['platform']}: "
+                           f"{entry['baseline']!r} -> {entry['candidate']!r}"
+                           for entry in platform_adjustments) + "\n")
+    (EVIDENCE / "summary.md").write_text(summary, encoding="utf-8")
+    with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as file:
+        file.write(summary)
+    print(summary)
+    return int(bool(problems))
+
+
+def main():
+    EVIDENCE.mkdir(exist_ok=True)
+    if sys.argv[1] == "build":
+        build()
+        return 0
+    if sys.argv[1] == "qualify":
+        return qualify()
+    if sys.argv[1] == "suite":
+        source, library, sha, version, test_file, label = sys.argv[2:]
+        return suite(Path(source), Path(library), sha, version, test_file, label)
+    raise ValueError("Unknown qualification mode")
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/scripts/test_fragment_offset_qualification.py b/scripts/test_fragment_offset_qualification.py
new file mode 100644
index 00000000..c65e8ac8
--- /dev/null
+++ b/scripts/test_fragment_offset_qualification.py
@@ -0,0 +1,305 @@
+"""Pure stdlib tests: never import c2pa or execute Cargo."""
+
+import ast
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import patch
+import xml.etree.ElementTree as ET
+
+import fragment_offset_qualification as q
+
+
+class QualificationTests(unittest.TestCase):
+    def test_compare_requires_every_common_test(self):
+        self.assertTrue(q.compare({"lost": ("passed", "", "")}, {}))
+
+    def test_compare_accepts_identical_baseline_failure_and_fixes(self):
+        failure = ("failure", "AssertionError", "expected output")
+        self.assertEqual(q.compare({"a": failure}, {"a": failure}), [])
+        self.assertEqual(q.compare({"a": failure}, {"a": ("passed", "", "")}), [])
+
+    def test_compare_rejects_new_or_changed_failure_type_message_and_skip(self):
+        failure = ("failure", "AssertionError", "expected output")
+        for outcome in (("failure", "ValueError", failure[2]),
+                        ("failure", failure[1], "different output"),
+                        ("skipped", "", "unsupported")):
+            with self.subTest(outcome=outcome):
+                self.assertTrue(q.compare({"a": failure}, {"a": outcome}))
+                self.assertTrue(q.compare({}, {"new": outcome}))
+
+    def test_windows_fork_applicability_is_exact_and_recorded(self):
+        skipped = ("skipped", "pytest.skip", "requires os.fork")
+        adjustments = []
+        with patch.object(q, "sys", SimpleNamespace(platform="win32")), patch.object(q, "os", SimpleNamespace()):
+            self.assertTrue(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE}, {q.FORK_TEST_ID: skipped}))
+            self.assertEqual(q.compare({q.FORK_TEST_ID: q.FORK_BASELINE_FAILURE},
+                                       {q.FORK_TEST_ID: skipped}, platform_adjustments=adjustments, context="native"), [])
+        self.assertEqual(adjustments, [{"test_id": q.FORK_TEST_ID, "platform": "win32",
+                                       "context": "native", "baseline": q.FORK_BASELINE_FAILURE, "candidate": skipped}])
+
+    def test_fork_applicability_does_not_relax_other_comparisons(self):
+        skipped = ("skipped", "pytest.skip", "requires os.fork")
+        for platform, os_module, key, previous, outcome in (
+                ("linux", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, skipped),
+                ("win32", SimpleNamespace(fork=object()), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, skipped),
+                ("win32", SimpleNamespace(), "other::test", q.FORK_BASELINE_FAILURE, skipped),
+                ("win32", SimpleNamespace(), q.FORK_TEST_ID, ("passed", "", ""), skipped),
+                ("win32", SimpleNamespace(), q.FORK_TEST_ID, ("failure", "builtins.AssertionError", "different failure"), skipped),
+                ("win32", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, ("skipped", "pytest.skip", "other reason")),
+                ("win32", SimpleNamespace(), q.FORK_TEST_ID, q.FORK_BASELINE_FAILURE, ("skipped", "other.type", "requires os.fork"))):
+            with self.subTest(platform=platform, key=key, previous=previous, outcome=outcome):
+                adjustments = []
+                with patch.object(q, "sys", SimpleNamespace(platform=platform)), patch.object(q, "os", os_module):
+                    self.assertTrue(q.compare({key: previous}, {key: outcome}, platform_adjustments=adjustments, context="native"))
+                self.assertEqual(adjustments, [])
+
+    def test_required_focused_ids_match_source_without_native_imports(self):
+        source = Path(q.__file__).resolve().parents[1] / "tests/test_fragment_at_offset.py"
+        tree = ast.parse(source.read_text(encoding="utf-8"))
+        ids = {f"tests.test_fragment_at_offset.{cls.name}::{method.name}"
+               for cls in tree.body if isinstance(cls, ast.ClassDef)
+               for method in cls.body if isinstance(method, ast.FunctionDef) and method.name.startswith("test_")}
+        self.assertEqual(len(q.REQUIRED_OFFSET_TEST_IDS), 18)
+        self.assertEqual(len(q.BINDING_OFFSET_TEST_IDS), 11)
+        self.assertEqual(len(q.NATIVE_OFFSET_TEST_IDS), 7)
+        self.assertLessEqual(q.REQUIRED_OFFSET_TEST_IDS, ids)
+
+    def test_focused_required_ids_cannot_be_replaced_at_constant_count(self):
+        report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", ""))
+        removed = sorted(q.REQUIRED_OFFSET_TEST_IDS)[0]
+        del report[removed]
+        report["unrelated.T::test_other"] = ("passed", "", "")
+        self.assertEqual(len(report), 18)
+        self.assertEqual(q.focused_problems(report), [f"Missing required focused test: {removed}"])
+
+    def test_focused_allows_extra_passes_but_no_extra_failure_or_skip(self):
+        report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", ""))
+        self.assertEqual(q.focused_problems(report), [])
+        report["future.T::test_extra"] = ("passed", "", "")
+        self.assertEqual(q.focused_problems(report), [])
+        for key in (sorted(q.REQUIRED_OFFSET_TEST_IDS)[0], "future.T::test_extra"):
+            for outcome in (("failure", "AssertionError", "bad"), ("skipped", "pytest.skip", "unsupported")):
+                with self.subTest(key=key, outcome=outcome):
+                    self.assertTrue(q.focused_problems({**report, key: outcome}))
+
+    def test_stock_focused_requires_binding_passes_and_exact_native_skips(self):
+        report = dict.fromkeys(q.BINDING_OFFSET_TEST_IDS, ("passed", "", ""))
+        report.update(dict.fromkeys(q.NATIVE_OFFSET_TEST_IDS, q.STOCK_NATIVE_SKIP))
+        self.assertEqual(q.focused_problems(report, stock=True), [])
+        self.assertEqual(q.focused_problems({**report, "future.T::test_extra": ("passed", "", "")}, stock=True), [])
+        native_id = sorted(q.NATIVE_OFFSET_TEST_IDS)[0]
+        binding_id = sorted(q.BINDING_OFFSET_TEST_IDS)[0]
+        for key, outcome in ((binding_id, q.STOCK_NATIVE_SKIP),
+                             (native_id, ("skipped", "pytest.skip", "wrong reason")),
+                             (native_id, ("skipped", "other.type", q.STOCK_NATIVE_SKIP[2])),
+                             (native_id, ("passed", "", "")),
+                             ("future.T::test_extra", q.STOCK_NATIVE_SKIP)):
+            with self.subTest(key=key, outcome=outcome):
+                self.assertTrue(q.focused_problems({**report, key: outcome}, stock=True))
+        del report[native_id]
+        report["unrelated.T::test_other"] = ("passed", "", "")
+        self.assertEqual(len(report), 18)
+        self.assertTrue(q.focused_problems(report, stock=True))
+
+    def test_fork_guard_is_per_test_and_preserves_unix_applicability(self):
+        source = Path(q.__file__).resolve().parents[1] / "tests/test_unit_tests_threaded.py"
+        tree = ast.parse(source.read_text(encoding="utf-8"))
+        cls = next(node for node in tree.body if isinstance(node, ast.ClassDef)
+                   and node.name == "TestForkedChildDoesNotDeadlock")
+        self.assertEqual(cls.decorator_list, [])
+        method = next(node for node in cls.body if isinstance(node, ast.FunctionDef)
+                      and node.name == "test_parent_copy_unaffected")
+        module = ast.Module(body=[method], type_ignores=[])
+        for supports_fork in (False, True):
+            namespace = {"unittest": unittest, "os": SimpleNamespace(**({"fork": object()} if supports_fork else {}))}
+            exec(compile(module, str(source), "exec"), namespace)
+            function = namespace[method.name]
+            self.assertEqual(getattr(function, "__unittest_skip__", False), not supports_fork)
+            if not supports_fork:
+                self.assertEqual(function.__unittest_skip_why__, "requires os.fork")
+
+    def test_normalize_only_paths_and_addresses(self):
+        with patch.object(q, "ROOT", Path("/work")):
+            self.assertEqual(q.normalize("/work/baseline/test.py 0x123abc vs 42"),
+                             q.normalize("/work/candidate/test.py 0x456def vs 42"))
+            self.assertNotEqual(q.normalize("expected 41"), q.normalize("expected 42"))
+        with patch.object(q.tempfile, "gettempdir", return_value="/temp"):
+            temporary = Path("/temp").resolve().as_posix()
+            self.assertEqual(q.normalize(temporary + "/tmp12_abcd3/file.jpg"), "<temporary>/file.jpg")
+            self.assertNotEqual(q.normalize("/temp/real-file.jpg"), "<temporary>")
+
+    def test_command_propagates_errors_and_only_defers_test_failures(self):
+        with tempfile.TemporaryDirectory() as directory, patch.object(q, "EVIDENCE", Path(directory)):
+            with patch.object(q.subprocess, "run", return_value=subprocess.CompletedProcess([], 1)) as run:
+                with self.assertRaises(subprocess.CalledProcessError):
+                    q.command(["fake"], directory, "build.log")
+                self.assertEqual(q.command(["fake"], directory, "test.log", test=True), 1)
+                self.assertEqual(run.call_args.kwargs["cwd"], directory)
+            with patch.object(q.subprocess, "run", return_value=subprocess.CompletedProcess([], 2)):
+                with self.assertRaises(subprocess.CalledProcessError):
+                    q.command(["fake"], directory, "test.log", test=True)
+
+    def test_no_local_native_build(self):
+        with patch.dict(os.environ, {}, clear=True), patch.object(q, "git") as git:
+            with self.assertRaises(RuntimeError):
+                q.validate_native(Path("unused"))
+            git.assert_not_called()
+
+    def test_build_cannot_execute_tools_before_validation(self):
+        with patch.object(q, "validate_native", side_effect=ValueError("refused")):
+            with patch.object(q, "command") as command:
+                with self.assertRaises(ValueError):
+                    q.build()
+                command.assert_not_called()
+
+    def test_native_allowlist_before_source_execution(self):
+        env = {"GITHUB_ACTIONS": "true", "GITHUB_REPOSITORY": "mstattma/c2pa-python"}
+        for origin, sha in (("https://github.com/other/c2pa-rs", q.NATIVE_SHA),
+                            (f"https://github.com/{q.NATIVE_REPOSITORY}", "f" * 40)):
+            with self.subTest(origin=origin, sha=sha), patch.dict(os.environ, env):
+                with patch.object(q, "git", side_effect=[origin, sha]):
+                    with self.assertRaises(ValueError):
+                        q.validate_native(Path("unused"))
+
+    def test_junit_ids_types_messages_and_empty_or_duplicate_rejection(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "report.xml"
+            xml = '<testcase classname="tests.T" name="test_x"><failure type="AssertionError" message="bad 0x123"/></testcase>'
+            path.write_text(f"<testsuite>{xml}</testsuite>", encoding="utf-8")
+            self.assertEqual(q.cases(path), {"tests.T::test_x": ("failure", "AssertionError", "bad 0xADDR")})
+            for body in ("", xml + xml):
+                path.write_text(f"<testsuite>{body}</testsuite>", encoding="utf-8")
+                with self.assertRaises(ValueError):
+                    q.cases(path)
+
+    def test_junit_requires_and_preserves_explicit_exception_type(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "report.xml"
+            for properties in ("", '<properties><property name="exception_type" value="builtins.AssertionError"/></properties>'):
+                path.write_text('<testsuite><testcase classname="T" name="test">'
+                                + properties + '<failure message="bad"/></testcase></testsuite>', encoding="utf-8")
+                if properties:
+                    self.assertEqual(q.cases(path)["T::test"], ("failure", "builtins.AssertionError", "bad"))
+                else:
+                    with self.assertRaises(ValueError):
+                        q.cases(path)
+
+    def test_qualification_keeps_full_failure_distinct_and_enforces_gates(self):
+        for scenario in ("unchanged", "native-failure", "native-offset-skip", "stock-failure", "stock-new-skip", "stock-missing-common",
+                         "replace-required-id", "stock-binding-skip", "stock-native-wrong-reason",
+                         "stock-baseline-unexpected-failure", "windows-fork"):
+            with self.subTest(scenario=scenario), tempfile.TemporaryDirectory() as directory:
+                root = Path(directory).resolve()
+                evidence = root / "evidence"
+                evidence.mkdir()
+                (root / "baseline").mkdir()
+                stock_dir = root / "candidate/artifacts/platform"
+                stock_dir.mkdir(parents=True)
+                stock = stock_dir / "native.bin"
+                stock.write_bytes(b"stock")
+                native = evidence / "native.bin"
+                native.write_bytes(b"candidate")
+                (evidence / "native.json").write_text(json.dumps({"library": str(native), "sha256": q.digest(native)}), encoding="utf-8")
+                seen = []
+
+                def fake_command(args, cwd, name, env=None, test=False):
+                    if not test:
+                        return 0
+                    label = args[-1]
+                    seen.append((label, args, env))
+                    focused = label.endswith("test_fragment_at_offset")
+                    if focused:
+                        report = dict.fromkeys(q.REQUIRED_OFFSET_TEST_IDS, ("passed", "", ""))
+                        if label.startswith("stock-"):
+                            report.update(dict.fromkeys(q.NATIVE_OFFSET_TEST_IDS, q.STOCK_NATIVE_SKIP))
+                        native_id = sorted(q.NATIVE_OFFSET_TEST_IDS)[0]
+                        if scenario == "native-offset-skip" and label.startswith("candidate-"):
+                            report[native_id] = q.STOCK_NATIVE_SKIP
+                        if scenario == "replace-required-id" and label.startswith("candidate-"):
+                            del report[native_id]
+                            report["unrelated.T::test_extra"] = ("passed", "", "")
+                            self.assertEqual(len(report), 18)
+                        if scenario == "stock-binding-skip" and label.startswith("stock-"):
+                            report[sorted(q.BINDING_OFFSET_TEST_IDS)[0]] = q.STOCK_NATIVE_SKIP
+                        if scenario == "stock-native-wrong-reason" and label.startswith("stock-"):
+                            report[native_id] = ("skipped", "pytest.skip", "wrong reason")
+                    else:
+                        report = {"T::test_common": ("passed", "", "")}
+                        if label.startswith(("baseline-", "candidate-")):
+                            message = "changed" if scenario == "native-failure" and label.startswith("candidate-") else "baseline failure"
+                            report["T::test_common"] = ("failure", "AssertionError", message)
+                        if label == "stock-test_unit_tests":
+                            if scenario == "stock-failure":
+                                report["T::test_common"] = ("failure", "AssertionError", "stock failure")
+                            if scenario == "stock-new-skip":
+                                report["T::test_common"] = ("skipped", "pytest.skip", "hidden regression")
+                            if scenario == "stock-missing-common":
+                                report = {"T::test_unrelated": ("passed", "", "")}
+                        if scenario == "stock-baseline-unexpected-failure" and label == "stock-baseline-test_unit_tests":
+                            report["T::test_common"] = ("failure", "AssertionError", "unexpected baseline failure")
+                        if scenario == "windows-fork" and label.endswith("test_unit_tests_threaded"):
+                            report[q.FORK_TEST_ID] = (q.FORK_BASELINE_FAILURE if label.startswith(("baseline-", "stock-baseline-"))
+                                                      else ("skipped", "pytest.skip", "requires os.fork"))
+                    xml = ET.Element("testsuite")
+                    for key, (kind, failure_type, message) in report.items():
+                        classname, name = key.split("::", 1)
+                        case = ET.SubElement(xml, "testcase", classname=classname, name=name)
+                        if kind != "passed":
+                            ET.SubElement(case, kind, type=failure_type, message=message)
+                    ET.ElementTree(xml).write(evidence / (label + ".xml"), encoding="utf-8")
+                    return int(any(outcome[0] in ("failure", "error") for outcome in report.values()))
+
+                with patch.object(q, "ROOT", root), patch.object(q, "EVIDENCE", evidence):
+                    with patch.object(q, "command", side_effect=fake_command):
+                        with patch.dict(os.environ, {"GITHUB_STEP_SUMMARY": str(root / "summary.md")}), patch("builtins.print"):
+                            sys_module = SimpleNamespace(platform="win32" if scenario == "windows-fork" else "linux", executable=q.sys.executable)
+                            with patch.object(q, "sys", sys_module), patch.object(q, "os", SimpleNamespace(environ=os.environ)):
+                                self.assertEqual(q.qualify(), int(scenario not in ("unchanged", "windows-fork")))
+                summary = (evidence / "summary.md").read_text(encoding="utf-8")
+                comparison = json.loads((evidence / "comparison.json").read_text(encoding="utf-8"))
+                if scenario == "windows-fork":
+                    self.assertEqual({entry["context"] for entry in comparison["platform_adjustments"]}, {"stock", "native"})
+                    self.assertEqual(len(comparison["platform_adjustments"]), 2)
+                    self.assertEqual(comparison["exit_codes"]["stock-baseline-test_unit_tests_threaded"], 1)
+                    self.assertEqual(comparison["exit_codes"]["stock-test_unit_tests_threaded"], 0)
+                else:
+                    self.assertEqual(comparison["platform_adjustments"], [])
+                if scenario in ("stock-new-skip", "stock-missing-common"):
+                    self.assertEqual(comparison["exit_codes"]["stock-test_unit_tests"], 0)
+                    self.assertTrue(comparison["comparisons"]["stock"])
+                self.assertIn("Candidate full-suite status: **FAIL**", summary)
+                self.assertIn(f"No-new-regressions: **{'FAIL' if scenario in ('native-failure', 'stock-failure', 'stock-new-skip', 'stock-missing-common') else 'PASS'}**", summary)
+                self.assertIn(f"Stock 0.91 compatibility: **{'FAIL' if scenario.startswith('stock-') else 'PASS'}**", summary)
+                self.assertEqual(len(seen), 10)
+                self.assertEqual(set(comparison["lanes"]), {"stock-baseline", "stock", "baseline", "candidate"})
+                for label, args, env in seen:
+                    if label.startswith("candidate-"):
+                        self.assertEqual(env["C2PA_FRAGMENT_OFFSET_REQUIRED"], "1")
+                    if label.startswith("stock-"):
+                        self.assertNotIn("C2PA_PREFLIGHT_RUN", env)
+                        self.assertNotIn("C2PA_FRAGMENT_OFFSET_REQUIRED", env)
+                        self.assertEqual(env["C2PA_LIBRARY_NAME"], str(stock))
+                        self.assertEqual(args[5], q.digest(stock))
+                    else:
+                        self.assertEqual(env["C2PA_PREFLIGHT_RUN"], "1")
+                        self.assertEqual(env["C2PA_LIBRARY_NAME"], str(native))
+                        self.assertEqual(args[5], q.digest(native))
+                    source = root / ("baseline" if label.startswith(("stock-baseline-", "baseline-")) else "candidate")
+                    self.assertEqual(args[3], str(source))
+                    self.assertEqual(env["PYTHONPATH"], str(source / "src"))
+
+    def test_loaded_provenance_rejects_fallback_without_native_imports(self):
+        fake = SimpleNamespace(_lib=SimpleNamespace(_name="relative-fallback.so"))
+        package = SimpleNamespace(c2pa=fake)
+        with patch.dict("sys.modules", {"c2pa": package, "c2pa.c2pa": fake, "pytest": SimpleNamespace()}):
+            with self.assertRaisesRegex(RuntimeError, "Wrong loaded native"):
+                q.suite(Path("source"), Path("/expected/library"), "hash", "version", "test.py", "test")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/src/c2pa/__init__.py b/src/c2pa/__init__.py
index bd581d58..961617af 100644
--- a/src/c2pa/__init__.py
+++ b/src/c2pa/__init__.py
@@ -32,7 +32,8 @@
     ContextBuilder,
     ContextProvider,
     sdk_version,
-    load_settings
+    load_settings,
+    has_fragment_at_offset
 )  # NOQA
 
 # Re-export C2paError and its subclasses
@@ -51,5 +52,6 @@
     'ContextBuilder',
     'ContextProvider',
     'sdk_version',
-    'load_settings'
+    'load_settings',
+    'has_fragment_at_offset'
 ]
diff --git a/src/c2pa/c2pa.py b/src/c2pa/c2pa.py
index eca9e952..5a91d873 100644
--- a/src/c2pa/c2pa.py
+++ b/src/c2pa/c2pa.py
@@ -168,6 +168,20 @@ def _validate_library_exports(lib):
 
 _validate_library_exports(_lib)
 
+# Offset-aware single-file fragment verification. Optional, so native
+# libraries that predate it still import; see has_fragment_at_offset().
+_FRAGMENT_AT_OFFSET_FUNCTIONS = (
+    'c2pa_reader_with_fragment_at_offset',
+    'c2pa_reader_fragment_at_offset_supported',
+)
+_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE = all(
+    hasattr(_lib, name) for name in _FRAGMENT_AT_OFFSET_FUNCTIONS
+)
+_FRAGMENT_OFFSET_MAX = 2**64 - 1
+# Cached probe result; None until has_fragment_at_offset() first runs, so
+# importing the module never calls into the native library for it.
+_fragment_at_offset_probe = None
+
 
 class C2paSeekMode(enum.IntEnum):
     """Seek mode for stream operations."""
@@ -1503,6 +1517,16 @@ def _setup_function(func, argtypes, restype=None):
      ctypes.POINTER(C2paStream), ctypes.POINTER(C2paStream)],
     ctypes.POINTER(C2paReader)
 )
+if _FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE:
+    _setup_function(
+        _lib.c2pa_reader_with_fragment_at_offset,
+        [ctypes.POINTER(C2paReader), ctypes.c_char_p,
+         ctypes.POINTER(C2paStream), ctypes.POINTER(C2paStream),
+         ctypes.c_uint64],
+        ctypes.POINTER(C2paReader)
+    )
+    _setup_function(
+        _lib.c2pa_reader_fragment_at_offset_supported, [], ctypes.c_bool)
 _setup_function(
     _lib.c2pa_builder_from_context,
     [ctypes.POINTER(C2paContext)],
@@ -1941,6 +1965,42 @@ def _to_utf8_bytes(data: Union[str, dict],
         raise C2paError.Encoding(f"Invalid UTF-8 in {error_context}: {e}")
 
 
+def has_fragment_at_offset() -> bool:
+    """Return whether the native library supports offset-aware fragment
+    verification (:meth:`Reader.with_fragment_at_offset`).
+
+    True only when the library exports both
+    ``c2pa_reader_with_fragment_at_offset`` and
+    ``c2pa_reader_fragment_at_offset_supported`` and the latter reports
+    support. The probe runs on first use, not at import, and is cached.
+    """
+    global _fragment_at_offset_probe
+    if not _FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE:
+        return False
+    if _fragment_at_offset_probe is None:
+        _fragment_at_offset_probe = bool(
+            _lib.c2pa_reader_fragment_at_offset_supported())
+    return _fragment_at_offset_probe
+
+
+def _require_fragment_at_offset() -> None:
+    if not has_fragment_at_offset():
+        raise C2paError.NotSupported(
+            "Offset-aware fragment verification is not available in the "
+            "loaded native library (c2pa_reader_with_fragment_at_offset)")
+
+
+def _check_fragment_offset(fragment_offset) -> int:
+    """Validate a fragment offset before any native work."""
+    if isinstance(fragment_offset, bool) or not isinstance(
+            fragment_offset, int):
+        raise TypeError("fragment_offset must be an int")
+    if not 0 <= fragment_offset <= _FRAGMENT_OFFSET_MAX:
+        raise ValueError(
+            f"fragment_offset must be between 0 and {_FRAGMENT_OFFSET_MAX}")
+    return fragment_offset
+
+
 def sdk_version() -> str:
     """
     Returns the underlying c2pa-rs/c2pa-c-ffi version string
@@ -2885,6 +2945,22 @@ def _encode_format(
             f"Invalid UTF-8 characters in input: {e}")
 
 
+def _fragment_offset_format_arg(
+    format: Optional[str], stream: Any, fragment_stream: Any,
+) -> bytes:
+    """Validate offset-aware fragment inputs before allocating native objects."""
+    if format is not None and not isinstance(format, str):
+        raise TypeError("format must be a str or None")
+    format_arg = _format_ffi_arg(_encode_format(format, "Reader"))
+    _check_cstr_arg('format', format_arg)
+    for name, value in (("stream", stream), ("fragment_stream", fragment_stream)):
+        if not Stream.is_read_stream(value):
+            raise TypeError(f"{name} must be a stream-like object")
+    if stream is fragment_stream:
+        raise ValueError("stream and fragment_stream must be distinct objects")
+    return format_arg
+
+
 def _format_ffi_arg(fmt: Optional[bytes]) -> bytes:
     """Convert an encoded format to the native FFI argument.
 
@@ -3402,7 +3478,142 @@ def with_fragment(self, format: Optional[str], stream,
         this call refused; yield between reads.
         """
         format_arg = _format_ffi_arg(_encode_format(format, "Reader"))
+        return self._swap_in_fragment(
+            format_arg, stream, fragment_stream,
+            lambda handle, main, frag: _lib.c2pa_reader_with_fragment(
+                handle, format_arg, main, frag))
+
+    def with_fragment_at_offset(self, format: Optional[str], stream,
+                                fragment_stream,
+                                fragment_offset: int) -> "Reader":
+        """Process one fragment of a single-file fragmented BMFF asset,
+        verified at its absolute byte offset in that asset.
+
+        A single-file fragmented MP4 hashes its fragments together with their
+        absolute root-box offsets, so a byte-range slice of it (for example an
+        HLS ``EXT-X-BYTERANGE`` segment) only verifies when the verifier knows
+        where the slice sat in the file. ``fragment_offset=0`` behaves like
+        :meth:`with_fragment`.
+
+        With the default exclusions, ``stream`` holds every byte before the
+        first C2PA ``merkle`` uuid box, and ``fragment_stream`` starts at a
+        ``merkle`` uuid box and ends right before the next one (or at the end
+        of the file). With a valid init and a parseable fragment, hash failures
+        caused by a wrong cut or offset are reported as
+        ``assertion.bmffHash.mismatch`` rather than raising. Init-loading or
+        fragment-parsing errors raise ``C2paError``. The offset is supplied by
+        the caller; it proves the bytes belong to the Merkle leaf they name,
+        not where they were fetched from.
+
+        Settings and trust come from this reader's Context.
+
+        Args:
+            format: MIME type of the media (e.g., "video/mp4"), handled as
+                in :meth:`with_fragment`.
+            stream: Stream-like object with the initialization bytes.
+            fragment_stream: Stream-like object with the fragment bytes.
+            fragment_offset: Absolute byte offset of the fragment's first
+                byte in the original asset, from 0 to 2**64 - 1.
+
+        Returns:
+            This reader instance, for method chaining.
+
+        Raises:
+            C2paError.NotSupported: If the native library lacks the API
+                (see :func:`has_fragment_at_offset`). The reader is untouched.
+            TypeError, ValueError: For an invalid offset, format type, or
+                streams. The reader is untouched.
+            C2paError: For a NUL byte or invalid UTF-8 in the format. The
+                reader is untouched.
+            C2paError: As :meth:`with_fragment`, including the reader being
+                closed after a native failure.
+        """
+        _require_fragment_at_offset()
+        offset = _check_fragment_offset(fragment_offset)
+        format_arg = _fragment_offset_format_arg(format, stream, fragment_stream)
+        return self._swap_in_fragment(
+            format_arg, stream, fragment_stream,
+            lambda handle, main, frag: (
+                _lib.c2pa_reader_with_fragment_at_offset(
+                    handle, format_arg, main, frag, offset)))
 
+    @classmethod
+    def from_fragment_at_offset(
+        cls,
+        format: Optional[str],
+        init_stream: Any,
+        fragment_stream: Any,
+        fragment_offset: int,
+        context: Optional['ContextProvider'] = None,
+    ) -> "Reader":
+        """Create a Reader for one fragment of a single-file fragmented
+        BMFF asset, verified at its absolute byte offset.
+
+        Unlike ``Reader(format, init_stream, context=...)`` followed by
+        :meth:`with_fragment_at_offset`, the init stream is not read and
+        validated on its own first. See :meth:`with_fragment_at_offset` for
+        the cut rules and result semantics.
+
+        Args:
+            format: MIME type of the media (e.g., "video/mp4").
+            init_stream: Stream-like object with the initialization bytes.
+            fragment_stream: Stream-like object with the fragment bytes.
+            fragment_offset: Absolute byte offset of the fragment's first
+                byte in the original asset, from 0 to 2**64 - 1.
+            context: Optional ContextProvider supplying settings and trust.
+                When omitted, a default Context is used.
+
+        Raises:
+            C2paError.NotSupported: If the native library lacks the API.
+            TypeError, ValueError: For an invalid offset, format type,
+                streams, or context type, before any native object is created.
+            C2paError: For a NUL byte or invalid UTF-8 in the format, or an
+                invalid context, before any native object is created.
+            C2paError: If the reader cannot be created or the fragment
+                cannot be processed.
+        """
+        _require_fragment_at_offset()
+        _check_fragment_offset(fragment_offset)
+        _fragment_offset_format_arg(format, init_stream, fragment_stream)
+        if context is not None:
+            if not isinstance(context, ContextProvider):
+                raise TypeError("context must be a ContextProvider")
+            if not context.is_valid:
+                raise C2paError("Context is not valid")
+
+        owned_context = Context() if context is None else None
+        active_context = context if context is not None else owned_context
+        try:
+            reader = object.__new__(cls)
+            ManagedResource.__init__(reader)
+            reader._init_attrs()
+            with _context_guard(active_context):
+                reader._create_and_activate(
+                    lambda: _lib.c2pa_reader_from_context(
+                        active_context.execution_context),
+                    Reader._ERROR_MESSAGES['reader_error'])
+            if context is not None:
+                # Pin a caller-supplied Context for the reader's lifetime.
+                reader._context = context
+            try:
+                reader.with_fragment_at_offset(
+                    format, init_stream, fragment_stream, fragment_offset)
+            except Exception:
+                reader.close()
+                raise
+            return reader
+        finally:
+            # The native reader holds its own reference to the context.
+            if owned_context is not None:
+                owned_context.close()
+
+    def _swap_in_fragment(self, format_arg, stream, fragment_stream,
+                          native_call) -> "Reader":
+        """Run a consuming fragment call and keep its streams.
+
+        ``native_call(handle, main_stream, fragment_stream)`` consumes the
+        reader handle and returns the replacement (NULL on failure).
+        """
         # A forked child cannot wait on a lock no surviving thread will
         # release, so it reports the same error _live_op_lock() does.
         if is_foreign_process(self):
@@ -3429,12 +3640,8 @@ def with_fragment_call(handle):
                     # and a reader must never be served them.
                     self._manifest_json_str_cache = None
                     self._manifest_data_cache = None
-                    return _lib.c2pa_reader_with_fragment(
-                        handle,
-                        format_arg,
-                        main_obj._stream,
-                        frag_obj._stream,
-                    )
+                    return native_call(
+                        handle, main_obj._stream, frag_obj._stream)
 
                 self._consume_and_swap(
                     with_fragment_call,
@@ -4896,5 +5103,6 @@ def ed25519_sign(data: bytes, private_key: str) -> bytes:
     'load_settings',
     'format_embeddable',
     'version',
-    'sdk_version'
+    'sdk_version',
+    'has_fragment_at_offset'
 ]
diff --git a/tests/test_fragment_at_offset.py b/tests/test_fragment_at_offset.py
new file mode 100644
index 00000000..9b8db68c
--- /dev/null
+++ b/tests/test_fragment_at_offset.py
@@ -0,0 +1,301 @@
+# Copyright 2026 Adobe. All rights reserved.
+# This file is licensed to you under the Apache License,
+# Version 2.0 (http://www.apache.org/licenses/LICENSE-2.0)
+# or the MIT license (http://opensource.org/licenses/MIT),
+# at your option.
+
+# Unless required by applicable law or agreed to in writing,
+# this software is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR REPRESENTATIONS OF ANY KIND, either express or
+# implied. See the LICENSE-MIT and LICENSE-APACHE files for the
+# specific language governing permissions and limitations under
+# each license.
+
+"""Tests for Reader.with_fragment_at_offset / from_fragment_at_offset.
+
+Two groups:
+
+* Binding behavior that does not need the new native entry point: feature
+  detection, argument validation, and that a refused call leaves the reader
+  untouched. These run against any native library.
+* Native behavior, against a library that exports
+  c2pa_reader_with_fragment_at_offset. They are skipped when the loaded
+  library lacks it, unless C2PA_FRAGMENT_OFFSET_REQUIRED=1, which turns the
+  skip into a failure for CI lanes that build the native library.
+
+The native tests use the signed multi-file DASH fixtures: a multi-file
+fragment verifies at offset 0 and is a hash mismatch at any other offset. The
+offset arithmetic on single-file layouts is covered by the c2pa-rs tests.
+"""
+
+import io
+import json
+import os
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+import c2pa
+import c2pa.c2pa as c2pa_module
+from c2pa import C2paError as Error, Context, Reader
+
+FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")
+INIT_PATH = os.path.join(FIXTURES_DIR, "dashinit.mp4")
+FRAGMENT_PATH = os.path.join(FIXTURES_DIR, "dash1.m4s")
+DEFAULT_TEST_FILE = os.path.join(FIXTURES_DIR, "C.jpg")
+BMFF_MISMATCH = "assertion.bmffHash.mismatch"
+REQUIRED = os.environ.get("C2PA_FRAGMENT_OFFSET_REQUIRED") == "1"
+
+
+def _read(path):
+    with open(path, "rb") as f:
+        return f.read()
+
+
+def _failure_codes(reader):
+    store = json.loads(reader.json())
+    return [s.get("code") for s in store.get("validation_status") or []]
+
+
+class TestFragmentAtOffsetFeatureDetection(unittest.TestCase):
+
+    def test_exported_from_package(self):
+        self.assertIs(c2pa.has_fragment_at_offset,
+                      c2pa_module.has_fragment_at_offset)
+        self.assertIn("has_fragment_at_offset", c2pa.__all__)
+
+    def test_false_without_symbols(self):
+        with patch.object(
+                c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", False):
+            self.assertFalse(c2pa.has_fragment_at_offset())
+
+    def test_probe_result_is_used_and_cached(self):
+        with patch.object(
+                c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", True), \
+                patch.object(c2pa_module, "_fragment_at_offset_probe", None), \
+                patch.object(c2pa_module, "_lib") as lib:
+            lib.c2pa_reader_fragment_at_offset_supported.return_value = False
+            self.assertFalse(c2pa.has_fragment_at_offset())
+            self.assertFalse(c2pa.has_fragment_at_offset())
+            lib.c2pa_reader_fragment_at_offset_supported.assert_called_once()
+
+    def test_unsupported_raises_before_touching_the_reader(self):
+        context = Context()
+        reader = Reader(DEFAULT_TEST_FILE, context=context)
+        before = reader.json()
+        with patch.object(
+                c2pa_module, "_FRAGMENT_AT_OFFSET_SYMBOLS_AVAILABLE", False):
+            with self.assertRaises(Error.NotSupported):
+                reader.with_fragment_at_offset(
+                    "video/mp4", io.BytesIO(b"init"), io.BytesIO(b"frag"), 0)
+            with self.assertRaises(Error.NotSupported):
+                Reader.from_fragment_at_offset(
+                    "video/mp4", io.BytesIO(b"init"), io.BytesIO(b"frag"), 0)
+        self.assertTrue(reader.is_valid)
+        self.assertEqual(reader.json(), before)
+        reader.close()
+        context.close()
+
+
+class TestFragmentAtOffsetValidation(unittest.TestCase):
+    """Invalid arguments are refused before any native allocation or mutation."""
+
+    def setUp(self):
+        self.patch = patch.object(
+            c2pa_module, "has_fragment_at_offset", return_value=True)
+        self.patch.start()
+        self.context = Context()
+        self.reader = Reader(DEFAULT_TEST_FILE, context=self.context)
+        self.before = self.reader.json()
+
+    def tearDown(self):
+        self.reader.close()
+        self.context.close()
+        self.patch.stop()
+
+    def assert_untouched(self):
+        self.assertTrue(self.reader.is_valid)
+        self.assertEqual(self.reader.json(), self.before)
+
+    def assert_invalid_inputs(self, error, format, init, fragment):
+        handle = self.reader._handle
+        streams = (self.reader._own_stream, self.reader._fragment_streams)
+        with patch.object(c2pa_module, "_lib") as lib:
+            with self.assertRaises(error):
+                Reader.from_fragment_at_offset(format, init, fragment, 0)
+            with self.assertRaises(error):
+                self.reader.with_fragment_at_offset(format, init, fragment, 0)
+            # Includes Context, Reader, and Stream allocation entry points.
+            self.assertEqual(lib.mock_calls, [])
+        self.assertIs(self.reader._handle, handle)
+        self.assertIs(self.reader._own_stream, streams[0])
+        self.assertIs(self.reader._fragment_streams, streams[1])
+        self.assert_untouched()
+
+    def test_rejects_invalid_streams_before_allocation(self):
+        for invalid in (None, "init.mp4", Path("init.mp4"), b"init", object()):
+            for position in (0, 1):
+                with self.subTest(invalid=invalid, position=position):
+                    streams = [io.BytesIO(b"i"), io.BytesIO(b"f")]
+                    streams[position] = invalid
+                    self.assert_invalid_inputs(TypeError, "video/mp4", *streams)
+
+    def test_rejects_aliased_streams_before_allocation(self):
+        stream = io.BytesIO(b"i")
+        self.assert_invalid_inputs(ValueError, "video/mp4", stream, stream)
+
+    def test_rejects_invalid_format_types_before_allocation(self):
+        for format in (False, 0, 1, b"video/mp4", [], {}, object()):
+            with self.subTest(format=format):
+                self.assert_invalid_inputs(
+                    TypeError, format, io.BytesIO(b"i"), io.BytesIO(b"f"))
+
+    def test_rejects_invalid_format_encoding_before_allocation(self):
+        for format in ("video/mp4\x00", "\x00video/mp4", "video/\ud800"):
+            with self.subTest(format=format):
+                self.assert_invalid_inputs(
+                    Error, format, io.BytesIO(b"i"), io.BytesIO(b"f"))
+
+    def test_rejects_non_int_offsets(self):
+        for offset in (True, False, 1.0, "1", None, b"\x01"):
+            with self.subTest(offset=offset):
+                with self.assertRaises(TypeError):
+                    self.reader.with_fragment_at_offset(
+                        "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"),
+                        offset)
+                with self.assertRaises(TypeError):
+                    Reader.from_fragment_at_offset(
+                        "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"),
+                        offset)
+        self.assert_untouched()
+
+    def test_rejects_out_of_range_offsets(self):
+        for offset in (-1, 2**64, 2**70):
+            with self.subTest(offset=offset):
+                with self.assertRaises(ValueError):
+                    self.reader.with_fragment_at_offset(
+                        "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"),
+                        offset)
+                with self.assertRaises(ValueError):
+                    Reader.from_fragment_at_offset(
+                        "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"),
+                        offset)
+        self.assert_untouched()
+
+    def test_from_rejects_invalid_context(self):
+        with self.assertRaises(TypeError):
+            Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), 0,
+                context=object())
+        closed = Context()
+        closed.close()
+        with self.assertRaises(Error):
+            Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(b"i"), io.BytesIO(b"f"), 0,
+                context=closed)
+
+
+@unittest.skipUnless(
+    c2pa.has_fragment_at_offset() or REQUIRED,
+    "native library lacks c2pa_reader_with_fragment_at_offset")
+class TestFragmentAtOffsetNative(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        if not c2pa.has_fragment_at_offset():
+            raise AssertionError(
+                "C2PA_FRAGMENT_OFFSET_REQUIRED=1 but the native library "
+                "lacks c2pa_reader_with_fragment_at_offset")
+        cls.init = _read(INIT_PATH)
+        cls.fragment = _read(FRAGMENT_PATH)
+
+    def setUp(self):
+        self.context = Context()
+
+    def tearDown(self):
+        self.context.close()
+
+    def with_fragment_codes(self):
+        with Reader("video/mp4", io.BytesIO(self.init),
+                    context=self.context) as reader:
+            reader.with_fragment("video/mp4", io.BytesIO(self.init),
+                                 io.BytesIO(self.fragment))
+            return _failure_codes(reader)
+
+    def test_zero_offset_matches_with_fragment(self):
+        expected = self.with_fragment_codes()
+        self.assertNotIn(BMFF_MISMATCH, expected)
+        with Reader("video/mp4", io.BytesIO(self.init),
+                    context=self.context) as reader:
+            result = reader.with_fragment_at_offset(
+                "video/mp4", io.BytesIO(self.init),
+                io.BytesIO(self.fragment), 0)
+            self.assertIs(result, reader)
+            self.assertEqual(_failure_codes(reader), expected)
+
+    def test_from_fragment_at_offset_zero(self):
+        expected = self.with_fragment_codes()
+        for context in (self.context, None):
+            with self.subTest(context=context):
+                with Reader.from_fragment_at_offset(
+                        "video/mp4", io.BytesIO(self.init),
+                        io.BytesIO(self.fragment), 0,
+                        context=context) as reader:
+                    self.assertEqual(_failure_codes(reader), expected)
+
+    def test_wrong_offset_is_a_mismatch_not_an_error(self):
+        for offset in (1, len(self.init), 2**32, 2**63):
+            with self.subTest(offset=offset):
+                with Reader.from_fragment_at_offset(
+                        "video/mp4", io.BytesIO(self.init),
+                        io.BytesIO(self.fragment), offset,
+                        context=self.context) as reader:
+                    self.assertIn(BMFF_MISMATCH, _failure_codes(reader))
+
+    def test_maximum_offset_never_verifies(self):
+        # Offset arithmetic overflows in the verifier: either an error or a
+        # failure, never a clean result.
+        try:
+            reader = Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(self.init),
+                io.BytesIO(self.fragment), 2**64 - 1, context=self.context)
+        except Error:
+            return
+        with reader:
+            self.assertTrue(_failure_codes(reader))
+
+    def test_replacing_the_fragment_refreshes_results(self):
+        with Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(self.init),
+                io.BytesIO(self.fragment), 1,
+                context=self.context) as reader:
+            self.assertIn(BMFF_MISMATCH, _failure_codes(reader))
+            reader.with_fragment_at_offset(
+                "video/mp4", io.BytesIO(self.init),
+                io.BytesIO(self.fragment), 0)
+            self.assertNotIn(BMFF_MISMATCH, _failure_codes(reader))
+
+    def test_native_failure_closes_the_reader(self):
+        reader = Reader.from_fragment_at_offset(
+            "video/mp4", io.BytesIO(self.init),
+            io.BytesIO(self.fragment), 0, context=self.context)
+        with self.assertRaises(Error):
+            reader.with_fragment_at_offset(
+                "video/mp4", io.BytesIO(b"\x00" * 16),
+                io.BytesIO(b"\x00" * 16), 0)
+        self.assertFalse(reader.is_valid)
+        with self.assertRaises(Error):
+            reader.json()
+        reader.close()
+
+    def test_from_fails_cleanly_without_a_manifest(self):
+        with self.assertRaises(Error):
+            Reader.from_fragment_at_offset(
+                "video/mp4", io.BytesIO(b"\x00" * 16),
+                io.BytesIO(b"\x00" * 16), 0, context=self.context)
+        # The caller's context stays usable.
+        self.assertTrue(self.context.is_valid)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_unit_tests.py b/tests/test_unit_tests.py
index dd3c3ec3..a09f6a7f 100644
--- a/tests/test_unit_tests.py
+++ b/tests/test_unit_tests.py
@@ -3745,58 +3745,57 @@ def test_builder_sign_with_ingredients_edit_intent(self):
             self.assertIsInstance(ingredients_manifest, list)
             self.assertEqual(len(ingredients_manifest), 2, "Should have exactly two ingredients")
 
-            # Verify the first ingredient is the one we added manually with componentOf relationship
-            first_ingredient = ingredients_manifest[0]
-            self.assertEqual(first_ingredient["title"], "Test Ingredient")
-            self.assertEqual(first_ingredient["format"], "image/jpeg")
-            self.assertIn("instance_id", first_ingredient)
-            self.assertIn("thumbnail", first_ingredient)
-            self.assertEqual(first_ingredient["thumbnail"]["format"], "image/jpeg")
-            self.assertIn("identifier", first_ingredient["thumbnail"])
-            self.assertEqual(first_ingredient["relationship"], "componentOf")
-            self.assertIn("label", first_ingredient)
-
-            # Verify the second ingredient is the auto-created parent with parentOf relationship
-            second_ingredient = ingredients_manifest[1]
+            # Created/gathered buckets deliberately reorder ingredients (c2pa-rs #2720, 309b99abe).
+            self.assertCountEqual(
+                [ingredient["relationship"] for ingredient in ingredients_manifest],
+                ["parentOf", "componentOf"],
+            )
+            component_ingredient = next(
+                ingredient for ingredient in ingredients_manifest
+                if ingredient["relationship"] == "componentOf"
+            )
+            parent_ingredient = next(
+                ingredient for ingredient in ingredients_manifest
+                if ingredient["relationship"] == "parentOf"
+            )
+            self.assertEqual(component_ingredient["title"], "Test Ingredient")
+            self.assertEqual(component_ingredient["format"], "image/jpeg")
+            self.assertIn("instance_id", component_ingredient)
+            self.assertIn("thumbnail", component_ingredient)
+            self.assertEqual(component_ingredient["thumbnail"]["format"], "image/jpeg")
+            self.assertIn("identifier", component_ingredient["thumbnail"])
+            self.assertEqual(component_ingredient["relationship"], "componentOf")
+            self.assertIn("label", component_ingredient)
+
             # Parent ingredient may not have a title field, or may have an empty one
-            self.assertEqual(second_ingredient["format"], "image/jpeg")
-            self.assertIn("instance_id", second_ingredient)
-            self.assertIn("thumbnail", second_ingredient)
-            self.assertEqual(second_ingredient["thumbnail"]["format"], "image/jpeg")
-            self.assertIn("identifier", second_ingredient["thumbnail"])
-            self.assertEqual(second_ingredient["relationship"], "parentOf")
-            self.assertIn("label", second_ingredient)
-
-            # Count ingredients with parentOf relationship - should be exactly one
-            parent_ingredients = [
-                ing for ing in ingredients_manifest
-                if ing.get("relationship") == "parentOf"
-            ]
-            self.assertEqual(len(parent_ingredients), 1, "Should have exactly one parentOf ingredient")
+            self.assertEqual(parent_ingredient["format"], "image/jpeg")
+            self.assertIn("instance_id", parent_ingredient)
+            self.assertIn("thumbnail", parent_ingredient)
+            self.assertEqual(parent_ingredient["thumbnail"]["format"], "image/jpeg")
+            self.assertIn("identifier", parent_ingredient["thumbnail"])
+            self.assertEqual(parent_ingredient["relationship"], "parentOf")
+            self.assertIn("label", parent_ingredient)
 
             # Check that assertions exist
             self.assertIn("assertions", active_manifest)
             assertions = active_manifest["assertions"]
 
-            # Find the actions assertion
-            actions_assertion = None
-            for assertion in assertions:
-                if assertion["label"] in ["c2pa.actions", "c2pa.actions.v2"]:
-                    actions_assertion = assertion
-                    break
-
-            self.assertIsNotNone(actions_assertion, "Should have c2pa.actions assertion")
+            actions_assertions = [
+                assertion for assertion in assertions
+                if assertion["label"].split("__", 1)[0] == "c2pa.actions.v2"
+            ]
+            self.assertTrue(actions_assertions, "Should have c2pa.actions assertion")
 
             # Verify exactly one c2pa.opened action exists for EDIT intent
-            actions = actions_assertion["data"]["actions"]
             opened_actions = [
-                action for action in actions
+                action for assertion in actions_assertions
+                for action in assertion["data"]["actions"]
                 if action["action"] == "c2pa.opened"
             ]
             self.assertEqual(len(opened_actions), 1, "Should have exactly one c2pa.opened action")
 
             # Verify the c2pa.opened action has the correct structure with parameters and ingredients
-            opened_action = opened_actions[0]
+            opened_action, = opened_actions
             self.assertIn("parameters", opened_action, "c2pa.opened action should have parameters")
             self.assertIn("ingredients", opened_action["parameters"], "parameters should have ingredients array")
             ingredients_params = opened_action["parameters"]["ingredients"]
@@ -3807,6 +3806,7 @@ def test_builder_sign_with_ingredients_edit_intent(self):
             for ingredient_ref in ingredients_params:
                 self.assertIn("url", ingredient_ref, "Ingredient reference should have url")
                 self.assertIn("hash", ingredient_ref, "Ingredient reference should have hash")
+                self.assertEqual(ingredient_ref["url"].rsplit("/", 1)[-1], parent_ingredient["label"])
 
         builder.close()
 
@@ -5182,7 +5182,7 @@ def test_builder_minimal_manifest_add_actions_and_sign_no_auto_add(self):
         load_settings('{"builder":{"actions":{"auto_placed_action":{"enabled":true},"auto_opened_action":{"enabled":true},"auto_created_action":{"enabled":true,"source_type":"http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation"}}}}')
 
     def test_builder_minimal_manifest_add_actions_and_sign_with_auto_add(self):
-        # For testing, remove auto-added actions
+        # Enable auto-added actions.
         load_settings('{"builder":{"actions":{"auto_placed_action":{"enabled":true},"auto_opened_action":{"enabled":true},"auto_created_action":{"enabled":true,"source_type":"http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation"}}}}')
 
         initial_manifest_definition = {
@@ -5219,36 +5219,36 @@ def test_builder_minimal_manifest_add_actions_and_sign_with_auto_add(self):
             self.assertIn("assertions", active_manifest)
             assertions = active_manifest["assertions"]
 
-            # Find the c2pa.actions.v2 assertion to look for what we added
-            actions_assertion = None
-            for assertion in assertions:
-                if assertion.get("label") == "c2pa.actions.v2":
-                    actions_assertion = assertion
-                    break
-
-            self.assertIsNotNone(actions_assertion)
-            self.assertIn("data", actions_assertion)
-            assertion_data = actions_assertion["data"]
-            # Verify the manifest now contains actions
-            self.assertIn("actions", assertion_data)
-            actions = assertion_data["actions"]
-            # Verify "c2pa.created" action exists anywhere in the actions array
-            created_action_found = False
-            for action in actions:
-                if action.get("action") == "c2pa.created":
-                    created_action_found = True
-                    break
-
-            self.assertTrue(created_action_found)
-
-            # Verify "c2pa.color_adjustments" action also exists in the same actions array
-            color_adjustments_found = False
-            for action in actions:
-                if action.get("action") == "c2pa.color_adjustments":
-                    color_adjustments_found = True
-                    break
-
-            self.assertTrue(color_adjustments_found)
+            # c2pa-rs #2720 (309b99abe) separates synthesized created actions from gathered caller actions.
+            actions_assertions = [
+                assertion for assertion in assertions
+                if assertion["label"].split("__", 1)[0] == "c2pa.actions.v2"
+            ]
+            self.assertTrue(actions_assertions)
+            actions = []
+            for assertion in actions_assertions:
+                self.assertIn("data", assertion)
+                self.assertIn("actions", assertion["data"])
+                actions.extend(enumerate(assertion["data"]["actions"]))
+
+            inception_actions = [
+                (index, action) for index, action in actions
+                if action["action"] in ("c2pa.created", "c2pa.opened")
+            ]
+            self.assertEqual(len(inception_actions), 1, "Should have exactly one inception action")
+            (created_index, created_action), = inception_actions
+            self.assertEqual(created_index, 0, "Inception must be the first action of its assertion")
+            self.assertEqual(created_action["action"], "c2pa.created")
+            self.assertEqual(created_action["digitalSourceType"],
+                             "http://cv.iptc.org/newscodes/digitalsourcetype/digitalCreation")
+
+            color_adjustments = [
+                action for _, action in actions
+                if action["action"] == "c2pa.color_adjustments"
+            ]
+            self.assertEqual(len(color_adjustments), 1, "Should have exactly one color adjustment")
+            color_adjustment, = color_adjustments
+            self.assertEqual(color_adjustment["parameters"]["name"], "brightnesscontrast")
 
         builder.close()
 
diff --git a/tests/test_unit_tests_threaded.py b/tests/test_unit_tests_threaded.py
index 2722ea59..20faf67e 100644
--- a/tests/test_unit_tests_threaded.py
+++ b/tests/test_unit_tests_threaded.py
@@ -315,6 +315,7 @@ def test_is_valid_false_for_inherited_object(self):
         reader = self._foreign_reader_with_lock_held()
         self.assertFalse(reader.is_valid)
 
+    @unittest.skipUnless(hasattr(os, "fork"), "requires os.fork")
     def test_parent_copy_unaffected(self):
         """The child closing its copy must leave the parent's usable.