@@ -1,46 +1,35 @@
 import asyncio
-import json
 import os
 import signal
 import subprocess
-import tempfile
-from collections.abc import Mapping
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from pathlib import Path
-from typing import Any, TypeVar, cast
+from typing import TypeVar
 
 from temporalio import activity
-from temporalio.converter import value_to_type
 from temporalio.exceptions import ApplicationError
 
 from .models import (
     MavenInspection,
     PublicationInput,
     ReleaseResult,
     candidate_workflow_id,
-    json_value,
     matches_maven_payload,
-    maven_artifacts,
     publication_queue,
 )
+from .publication import PublicationFailure, Publisher
 
-T = TypeVar("T")
 PASSTHROUGH_ENV = {"PATH", "HOME", "TMPDIR", "LANG", "LC_ALL", "JAVA_HOME", "CI"}
-COMMAND_FAILURES = {
-    42: ("Immutable external identity or checksum conflict.", "ReleaseIdentityConflict"),
-    44: ("Maven repository identity is not yet visible.", "MavenSubmissionAmbiguous"),
-    45: ("Portal deployment failed validation.", "MavenDeploymentFailed"),
-    46: ("GitHub artifact expired or was deleted.", "ArtifactUnavailable"),
-}
+Result = TypeVar("Result", ReleaseResult, MavenInspection)
 
 
-async def run_trusted_command(cwd: Path, script: Path, env: Mapping[str, str]) -> list[str]:
-    """Run trusted shell reconciliation with an explicit, credential-minimal environment.
+async def run_trusted_process(cwd: Path, command: Sequence[str], env: Mapping[str, str]) -> bytes:
+    """Run a trusted process with an explicit, credential-minimal environment.
 
     The child gets only a small host-runtime allowlist plus values selected by the
     Activity. It runs in its own process group so Temporal cancellation can terminate
     the complete command tree, including curl, gh, Docker, Gradle, or GPG children.
     """
-    command = ["bash", str(script.absolute())]
     child_env = {key: value for key, value in os.environ.items() if key in PASSTHROUGH_ENV}
     child_env.update(env)
     process = await asyncio.create_subprocess_exec(
@@ -68,7 +57,13 @@ async def heartbeat() -> None:
         await asyncio.gather(beat, return_exceptions=True)
     if process.returncode:
         raise subprocess.CalledProcessError(process.returncode, command)
-    return stdout.decode().splitlines()
+    return stdout
+
+
+async def run_trusted_command(cwd: Path, script: Path, env: Mapping[str, str]) -> list[str]:
+    """Run a trusted shell script; retained as the small process-isolation test seam."""
+    output = await run_trusted_process(cwd, ["bash", str(script.absolute())], env)
+    return output.decode().splitlines()
 
 
 async def _terminate_tree(process: asyncio.subprocess.Process) -> None:
@@ -86,30 +81,33 @@ async def _terminate_tree(process: asyncio.subprocess.Process) -> None:
 class PublicationActivities:
     def __init__(
         self,
-        trusted_root: Path,
         source_root: Path,
         environment: Mapping[str, str],
     ) -> None:
-        """Capture trusted/source roots and a snapshot of the privileged job environment."""
-        self.trusted_root, self.source_root = trusted_root, source_root
+        """Capture the source root and a snapshot of the privileged job environment."""
+        self.source_root = source_root
         self.environment = dict(environment)
 
     @activity.defn(name="publishRelease")
     async def publish_release(self, value: PublicationInput) -> ReleaseResult:
         """Reconcile Maven and GitHub to the exact public release state."""
-        return await self._command(value, "all", ReleaseResult)
+        return await self._run(value, Publisher.publish_release)
 
     @activity.defn(name="inspectMaven")
     async def inspect_maven(self, value: PublicationInput) -> MavenInspection:
         """Inspect every durable Maven generation without creating or publishing state."""
-        return await self._command(value, "inspect", MavenInspection)
-
-    async def _command(self, value: PublicationInput, stage: str, result_type: type[T]) -> T:
-        """Validate Activity routing, execute reconciliation, and decode its typed result.
+        return await self._run(value, Publisher.inspect_maven)
 
-        Semantic conflicts and Maven ambiguity use stable exit codes that become typed,
-        non-retryable ApplicationErrors. Transient process and network failures remain
-        ordinary Activity failures and therefore use Temporal's configured retry policy.
+    async def _run(
+        self,
+        value: PublicationInput,
+        operation: Callable[[Publisher], Awaitable[Result]],
+    ) -> Result:
+        """Validate routing and execute one typed publication operation.
+
+        Durable semantic outcomes become non-retryable ApplicationErrors. Network,
+        process, and visibility failures remain ordinary exceptions so Temporal applies
+        the Workflow's retry policy.
         """
         try:
             expected_commit = self.environment.get("EXPECTED_COMMIT")
@@ -121,39 +119,20 @@ async def _command(self, value: PublicationInput, stage: str, result_type: type[
             raise ApplicationError(
                 str(error), type="InvalidPublicationInput", non_retryable=True
             ) from error
-        with tempfile.TemporaryDirectory(prefix="temporal-release-") as directory:
-            input_file, output_file = (
-                Path(directory) / name for name in ("input.json", "output.json")
-            )
-            data = json_value(value) | {
-                "releaseDigest": value.release.digest(),
-                "mavenArtifacts": maven_artifacts(value.release.candidate.mavenPolicy),
-            }
-            input_file.write_text(json.dumps(data, separators=(",", ":")))
-            env = {
-                "RELEASE_INPUT_FILE": str(input_file),
-                "RELEASE_OUTPUT_FILE": str(output_file),
-                "RELEASE_STAGE": stage,
-                "TRUSTED_AUTOMATION_ROOT": str(self.trusted_root),
-            }
-            for name in ("GH_TOKEN", "RH_USER", "RH_PASSWORD"):
-                if self.environment.get(name):
-                    env[name] = self.environment[name]
-            try:
-                output = await run_trusted_command(
-                    self.source_root,
-                    self.trusted_root / ".github/scripts/temporal-release/reconcile-publication.sh",
-                    env,
-                )
-            except subprocess.CalledProcessError as error:
-                if error.returncode not in COMMAND_FAILURES:
-                    raise
-                message, kind = COMMAND_FAILURES[error.returncode]
-                raise ApplicationError(message, type=kind, non_retryable=True) from error
-            if output:
-                raise RuntimeError("Publication command wrote unexpected standard output.")
-            raw: Any = json.loads(output_file.read_text())
-            return cast(T, value_to_type(result_type, raw))
+
+        async def command(arguments: Sequence[str], env: Mapping[str, str]) -> bytes:
+            """Run publication tools through the Activity's cancellation-aware boundary."""
+            return await run_trusted_process(self.source_root, arguments, env)
+
+        publisher: Publisher | None = None
+        try:
+            publisher = Publisher(value, self.source_root, self.environment, command)
+            return await operation(publisher)
+        except PublicationFailure as error:
+            raise ApplicationError(str(error), type=error.error_type, non_retryable=True) from error
+        finally:
+            if publisher is not None:
+                publisher.close()
 
 
 def validate_publication(
 
@@ -5,6 +5,7 @@
 import subprocess
 import sys
 from collections.abc import Mapping
+from contextlib import AsyncExitStack
 from pathlib import Path
 from typing import Any, TypeVar, cast
 
@@ -192,6 +193,14 @@ def workflow_queue(workflow_id: str) -> str:
     return candidate_queue_from_digest(workflow_id.removeprefix(prefix))
 
 
+def publication_worker_queues(queue: str) -> tuple[str, ...]:
+    """Return the current publication queue and its one bounded recovery successor."""
+    match = re.fullmatch(r"(sdk-java-release-[0-9a-f]{32}-publication-g)([01])", queue)
+    if match is None:
+        raise ValueError("Refusing to poll a non-publication Task Queue.")
+    return (queue, f"{match.group(1)}1") if match.group(2) == "0" else (queue,)
+
+
 def verify_candidate_origin(candidate: CandidateIdentity, run: Mapping[str, Any]) -> None:
     """Reconfirm that GitHub authorized the candidate from the expected push workflow.
 
@@ -293,33 +302,33 @@ async def output_jobs(client: Client, workflow_id: str, run_id: str) -> None:
 
 
 async def run_worker(client: Client, queue: str, env: Mapping[str, str]) -> None:
-    """Poll one release Workflow queue and one exact publication generation queue.
+    """Poll one release Workflow queue and the bounded publication generation queues.
 
     Hosting both Workers in the publication job removes a separate Actions worker
     matrix. Activities still exist only on the privileged generation-specific queue,
     while deterministic Workflow code polls the candidate-specific queue.
     """
-    if not queue.startswith("sdk-java-release-") or "-publication-g" not in queue:
-        raise ValueError("Refusing to poll a non-publication Task Queue.")
+    queues = publication_worker_queues(queue)
     workflow_id = required(env, "EXPECTED_WORKFLOW_ID")
     source = Path(required(env, "RELEASE_SOURCE_DIR")).absolute()
     if not source.is_dir():
         raise ValueError("RELEASE_SOURCE_DIR is not a directory.")
-    activities = PublicationActivities(ROOT, source, env)
+    activities = PublicationActivities(source, env)
     release_worker = Worker(
         client, task_queue=workflow_queue(workflow_id), workflows=[ReleaseWorkflow]
     )
-    publication_worker = Worker(
-        client,
-        task_queue=queue,
-        activities=[activities.publish_release, activities.inspect_maven],
-    )
     handle = client.get_workflow_handle(workflow_id, run_id=env.get("EXPECTED_RUN_ID"))
-    async with release_worker, publication_worker:
-        try:
-            await asyncio.wait_for(asyncio.shield(handle.result()), 95 * 60)
-        except TimeoutError:
-            pass
+    async with AsyncExitStack() as stack:
+        await stack.enter_async_context(release_worker)
+        for task_queue in queues:
+            await stack.enter_async_context(
+                Worker(
+                    client,
+                    task_queue=task_queue,
+                    activities=[activities.publish_release, activities.inspect_maven],
+                )
+            )
+        await handle.result()
 
 
 async def async_main(argv: list[str], env: Mapping[str, str]) -> None:
@@ -357,7 +366,7 @@ async def async_main(argv: list[str], env: Mapping[str, str]) -> None:
 
 def main() -> None:
     """Validate the trusted checkout layout and run the asynchronous CLI."""
-    if not (ROOT / ".github/scripts/temporal-release/reconcile-publication.sh").is_file():
+    if not (ROOT / ".github/scripts/temporal-release/prepare-maven-payload.sh").is_file():
         raise RuntimeError("The trusted repository root has an unexpected layout.")
     asyncio.run(async_main(sys.argv[1:], os.environ))
 
 
@@ -6,12 +6,13 @@
 import sys
 import tarfile
 import xml.etree.ElementTree as ET
+from collections.abc import Iterable
 
 
 def validate(
     root: pathlib.Path,
     manifest: pathlib.Path,
-    policy: pathlib.Path,
+    policy: Iterable[str],
     version: str,
     commit: str,
     exact: bool,
@@ -23,7 +24,7 @@ def validate(
     complete, preventing a partially generated payload from becoming durable release data.
     """
     root = root.resolve()
-    approved = set(policy.read_text().splitlines())
+    approved = set(policy)
     records = []
     for line in manifest.read_text().splitlines():
         relative, digest, size = line.split("\t")
@@ -122,7 +123,8 @@ def main() -> None:
         manifest = pathlib.Path(source)
     else:
         raise ValueError("expected extract or validate")
-    validate(root_path, manifest, pathlib.Path(policy), version, commit, mode == "validate")
+    approved = pathlib.Path(policy).read_text().splitlines()
+    validate(root_path, manifest, approved, version, commit, mode == "validate")
 
 
 if __name__ == "__main__":
 
@@ -0,0 +1,51 @@
+import gzip
+import io
+import stat
+import sys
+import tarfile
+import zipfile
+from pathlib import Path
+
+
+def package(source: Path, output: Path, root: str, name: str, windows: bool) -> None:
+    """Create a byte-reproducible native archive with one executable entry.
+
+    Fixed timestamps, ownership, permissions, and member order make two builds of
+    identical native bytes produce the same public archive on every runner.
+    """
+    data = source.read_bytes()
+    if windows:
+        entry = zipfile.ZipInfo(f"{root}/{name}", (1980, 1, 1, 0, 0, 0))
+        entry.compress_type = zipfile.ZIP_DEFLATED
+        entry.external_attr = (stat.S_IFREG | 0o755) << 16
+        with zipfile.ZipFile(output, "w") as archive:
+            archive.writestr(entry, data)
+        return
+
+    tar_bytes = io.BytesIO()
+    with tarfile.open(fileobj=tar_bytes, mode="w", format=tarfile.GNU_FORMAT) as archive:
+        directory = tarfile.TarInfo(root)
+        directory.type, directory.mode = tarfile.DIRTYPE, 0o755
+        directory.uid = directory.gid = directory.mtime = 0
+        directory.uname = directory.gname = ""
+        archive.addfile(directory)
+        binary = tarfile.TarInfo(f"{root}/{name}")
+        binary.size, binary.mode = len(data), 0o755
+        binary.uid = binary.gid = binary.mtime = 0
+        binary.uname = binary.gname = ""
+        archive.addfile(binary, io.BytesIO(data))
+    with (
+        output.open("wb") as raw,
+        gzip.GzipFile(filename="", mode="wb", fileobj=raw, mtime=0) as compressed,
+    ):
+        compressed.write(tar_bytes.getvalue())
+
+
+def main() -> None:
+    """Package the validated shell inputs without embedding Python in the script."""
+    source, output, root, name, platform = sys.argv[1:]
+    package(Path(source), Path(output), root, name, platform == "windows-amd64")
+
+
+if __name__ == "__main__":
+    main()