Verify shipped activity identity and document local retries

.github/workflows/ci.yml

@@ -19,6 +19,10 @@ on:
         description: Optional exact public Native source overlay SHA for candidate qualification
         type: string
         default: ""
+      published_activity_acceptance:
+        description: Run the reviewed published SDK/Server activity identity tuple
+        type: boolean
+        default: false
 
 permissions:
   contents: read
@@ -28,6 +32,31 @@ concurrency:
   cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
+  published-activity-acceptance:
+    if: github.event_name == 'workflow_dispatch' && inputs.published_activity_acceptance
+    needs: [action-policy, lint]
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@d23441a48e516b6c34aea4fa41551a30e30af803  # v6
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1  # v6
+        with:
+          python-version: "3.12.16"
+      - name: Bootstrap patched pip trust
+        uses: durable-workflow/.github/.github/actions/bootstrap-patched-pip@75587fc33ff40c34f32ef777293dad56bb6f48c7  # clean-trust
+        with:
+          clients: pip,python
+      - name: Run the published activity identity case
+        run: python scripts/ci/published-activity-acceptance.py
+      - name: Retain published tuple and scenario evidence
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7
+        with:
+          name: published-activity-acceptance-${{ github.sha }}
+          path: published-activity-results/
+          retention-days: 7
   action-policy:
     name: Central action policy preflight
     if: ${{ github.server_url == 'https://github.com' }}
@@ -104,7 +133,9 @@ jobs:
         with:
           clients: pip,python
       - run: pip install -e '.[dev]'
-      - run: ruff check src/ tests/
+      - run: >-
+          ruff check src/ tests/ scripts/ci/activity_acceptance_*.py
+          scripts/ci/verify_activity_install.py scripts/ci/published-activity-acceptance.py
       - run: mypy src/durable_workflow/
       - name: Enforce load-calibrated Avro Value regression budget
         run: python benchmarks/avro_value.py --enforce

docs/sdk-reference.md

@@ -268,7 +268,7 @@ receipt = yield ctx.start_child_workflow(
 )
 ```
 
-## Activity execution identity (unreleased)
+## Activity execution identity
 
 Inside an activity, `activity.context().info.activity_execution_id` exposes the
 logical execution ID received from Server. `workflow_id` and `run_id` identify
@@ -310,9 +310,8 @@ validated admission receipt's execution ID. Inline local activities have no
 admitted execution ID and expose `None`, while retaining their parent workflow
 and run IDs. Server-free mocks do not create Server activity identity.
 
-This source addition is unreleased and is outside the current typed-value
-release candidate. Published-package retry identity and execution-guarantees
-guide acceptance remain tracked in
+This accessor is available in Python SDK 2.15.0 and later. Published-Server retry
+identity and execution-guarantees guide acceptance remain tracked in
 [the shared identity issue](https://github.com/durable-workflow/.github/issues/211).
 
 ## Local activities
@@ -381,6 +380,13 @@ local attempts short; use `ctx.schedule_activity(...)` for separately queued
 work. Local activities are yielded one at a time; they are not members of a
 parallel workflow group.
 
+For local activities, `max_attempts` counts all attempts, including the first.
+When a nonempty `backoff_seconds` list runs out, later retries repeat its final
+entry. For example, five attempts with `[1, 2]` wait `1, 2, 2, 2` seconds before
+attempts two through five. An empty list adds no retry wait. A committed local
+result replays its recorded attempts without rerunning the handler, including
+older records made with a zero-delay tail.
+
 ## Worker sessions
 
 A worker session routes related activities to one capable worker while its

scripts/ci/activity_acceptance_consumer.py

@@ -0,0 +1,266 @@
+"""Install a closed wheel set, then await admission to the ordinary retry case."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import importlib.machinery
+import importlib.metadata
+import ipaddress
+import json
+import os
+import stat
+import subprocess
+import sys
+import time
+import urllib.request
+from pathlib import Path
+from urllib.parse import urlunsplit
+
+DATA = Path("/run/acceptance-data")
+WHEELS = Path("/run/acceptance-wheels")
+ENVIRONMENT = Path("/run/acceptance-env")
+LIMIT = 64 * 1024 * 1024
+JUNIT = DATA / "identity.xml"
+CAPTURED_VERIFIER = ""
+
+
+def receive(maximum):
+    data = sys.stdin.buffer.readline(maximum + 1)
+    if not data.endswith(b"\n") or len(data) > maximum:
+        raise ValueError("consumer input exceeds its bound")
+    return json.loads(data)
+
+
+def write(directory, name, item):
+    if not name or "/" in name or name in {".", ".."}:
+        raise ValueError("consumer input filename differs")
+    data = base64.b64decode(item["base64"], validate=True)
+    if hashlib.sha256(data).hexdigest() != item["sha256"]:
+        raise ValueError("consumer input digest differs")
+    descriptor = os.open(directory / name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
+    try:
+        with os.fdopen(descriptor, "wb", closefd=False) as stream:
+            stream.write(data)
+    finally:
+        os.close(descriptor)
+
+
+def run(arguments, environment, input_data=None):
+    return subprocess.run(
+        arguments, env=environment, input=input_data, capture_output=True, text=True, timeout=90, check=True
+    ).stdout
+
+
+def validate_setup(packet):
+    if set(packet) != {"files", "wheels", "wheel_name", "wheel_sha256", "sdk_version"}:
+        raise ValueError("consumer setup shape differs")
+    expected_files = {"requirements.txt", "verify_activity_install.py", "test_activity_identity.py", "conftest.py"}
+    if set(packet["files"]) != expected_files or not 1 <= len(packet["wheels"]) <= 30:
+        raise ValueError("consumer setup membership differs")
+    if packet["wheel_name"] not in packet["wheels"]:
+        raise ValueError("approved SDK wheel is missing")
+    if packet["wheels"][packet["wheel_name"]]["sha256"] != packet["wheel_sha256"]:
+        raise ValueError("SDK wheel packet differs from approval")
+
+
+def await_runtime(number):
+    if type(number) is not int or not 0 <= number <= 0xFFFFFFFF:
+        raise ValueError("consumer endpoint must be an unsigned IPv4 integer")
+    # The controller binds this numeric endpoint to the two inspected IDs.
+    address = ipaddress.IPv4Address(number)
+    if (
+        not address.is_private
+        or address.is_loopback
+        or address.is_unspecified
+        or address.is_multicast
+        or address.is_reserved
+        or address.is_link_local
+    ):
+        raise ValueError("consumer endpoint must be a private numeric IPv4 address")
+    octets = [(number >> shift) & 255 for shift in (24, 16, 8, 0)]
+    numeric_host = ".".join(str(octet) for octet in octets)
+    url = urlunsplit(("http", numeric_host + ":8080", "", "", ""))
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    deadline = time.monotonic() + 30
+    while time.monotonic() < deadline:
+        try:
+            with opener.open(url + "/api/health", timeout=2) as response:
+                if response.status == 200:
+                    return url
+        except OSError:
+            time.sleep(0.2)
+    raise TimeoutError("published runtime readiness")
+
+
+def retained_verifier():
+    if not CAPTURED_VERIFIER:
+        raise ValueError("controller did not bind the Git-captured verifier")
+    namespace = {"__name__": "captured_sdk_inventory"}
+    exec(compile(CAPTURED_VERIFIER, "<captured SDK inventory>", "exec"), namespace)
+    return namespace["verify_sdk"]
+
+
+def inventory(verifier, python, packet):
+    # The base interpreter, captured verifier and installer template remain in
+    # bootstrap memory. SDK execution cannot replace the terminal checker file.
+    from pip._vendor.distlib.scripts import SCRIPT_TEMPLATE
+
+    site = (
+        ENVIRONMENT
+        / "lib"
+        / ("python" + str(sys.version_info.major) + "." + str(sys.version_info.minor))
+        / "site-packages"
+    )
+    spec = importlib.machinery.PathFinder.find_spec("durable_workflow", [str(site)])
+    if spec is None or spec.origin is None:
+        raise ValueError("installed SDK origin is unavailable")
+    wheel = base64.b64decode(packet["wheels"][packet["wheel_name"]]["base64"], validate=True)
+    result = verifier(site, spec.origin, wheel, packet["wheel_sha256"], python=python, script_template=SCRIPT_TEMPLATE)
+    versions = [
+        item.version
+        for item in importlib.metadata.distributions(path=[str(site)])
+        if item.metadata["Name"].lower().replace("_", "-") == "durable-workflow"
+    ]
+    if versions != [packet["sdk_version"]]:
+        raise ValueError("installed SDK version differs")
+    return result
+
+
+def main():
+    packet = prepare_inputs()
+    verifier = retained_verifier()
+    environment, python = prepare_environment()
+    before = inventory(verifier, python, packet)
+    ready = {
+        "sdk_inventory": before,
+        "python": sys.version,
+        "environment": str(ENVIRONMENT),
+        "distributions": run(
+            [
+                str(python),
+                "-I",
+                "-B",
+                "-c",
+                "import importlib.metadata,json; print(json.dumps(sorted((d.metadata['Name'],"
+                "d.version) for d in importlib.metadata.distributions())))",
+            ],
+            environment,
+        ),
+    }
+    print("ACTIVITY_CONSUMER_READY=" + json.dumps(ready, sort_keys=True), flush=True)
+    admission = receive(8192)
+    if set(admission) != {"runtime_ipv4", "token"}:
+        raise ValueError("consumer admission shape differs")
+    admission["server_url"] = await_runtime(admission["runtime_ipv4"])
+    output = execute_case(verifier, python, packet, before, environment, admission)
+    print("ACTIVITY_CONSUMER_RESULT=" + json.dumps(output, sort_keys=True), flush=True)
+    # Remain attached while the controller rechecks the same two IDs and IP.
+    if receive(8192) != {"release": True}:
+        raise ValueError("consumer release shape differs")
+    if not output["complete"]:
+        raise RuntimeError("ordinary activity case failed")
+
+
+def prepare_inputs():
+    if os.getuid() != 1000:
+        raise ValueError("consumer must be unprivileged")
+    for directory in (DATA, WHEELS, ENVIRONMENT):
+        info = directory.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != 1000 or info.st_mode & 0o077 or list(directory.iterdir()):
+            raise ValueError("consumer scratch is not empty and private")
+    packet = receive(LIMIT)
+    validate_setup(packet)
+    for name, item in packet["files"].items():
+        write(DATA, name, item)
+    for name, item in packet["wheels"].items():
+        if not name.endswith(".whl"):
+            raise ValueError("consumer accepts wheels only")
+        write(WHEELS, name, item)
+    return packet
+
+
+def prepare_environment():
+    environment = {
+        "PATH": os.defpath,
+        "PYTHONDONTWRITEBYTECODE": "1",
+        "PYTHONNOUSERSITE": "1",
+        "PIP_CONFIG_FILE": os.devnull,
+        "PIP_DISABLE_PIP_VERSION_CHECK": "1",
+        "PIP_NO_CACHE_DIR": "1",
+        "PIP_NO_INPUT": "1",
+        "TMPDIR": str(DATA),
+    }
+    python = ENVIRONMENT / "bin/python"
+    run([sys.executable, "-I", "-B", "-m", "venv", "--without-pip", "--copies", str(ENVIRONMENT)], environment)
+    run(
+        [
+            sys.executable,
+            "-I",
+            "-B",
+            "-m",
+            "pip",
+            "--python",
+            str(python),
+            "install",
+            "--no-index",
+            "--find-links",
+            str(WHEELS),
+            "--only-binary=:all:",
+            "--require-hashes",
+            "--no-compile",
+            "-r",
+            str(DATA / "requirements.txt"),
+        ],
+        environment,
+    )
+    return environment, python
+
+
+def execute_case(verifier, python, packet, before, environment, admission):
+    environment.update(
+        {
+            "DURABLE_WORKFLOW_SERVER_URL": admission["server_url"],
+            "DURABLE_WORKFLOW_AUTH_TOKEN": admission["token"],
+            "DURABLE_WORKFLOW_EXPECT_INSTALLED_VERSION": packet["sdk_version"],
+            "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
+        }
+    )
+    result = subprocess.run(
+        [
+            str(python),
+            "-I",
+            "-B",
+            "-m",
+            "pytest",
+            "-c",
+            "/dev/null",
+            "-p",
+            "pytest_asyncio.plugin",
+            "-p",
+            "no:cacheprovider",
+            "--asyncio-mode=auto",
+            "--import-mode=importlib",
+            "-v",
+            "--capture=tee-sys",
+            str(DATA / "test_activity_identity.py"),
+            "--junitxml=" + str(JUNIT),
+        ],
+        env=environment,
+        capture_output=True,
+        text=True,
+        timeout=90,
+    )
+    after = inventory(verifier, python, packet)
+    if after != before:
+        raise ValueError("installed SDK inventory changed")
+    return {
+        "complete": result.returncode == 0,
+        "sdk_inventory": after,
+        "case_output": result.stdout + result.stderr,
+        "junit": JUNIT.read_text() if JUNIT.is_file() else "",
+    }
+
+
+if __name__ == "__main__":
+    main()