diff --git a/docs/sinnixd.md b/docs/sinnixd.md
index 11f36adc..8c6e62e1 100644
--- a/docs/sinnixd.md
+++ b/docs/sinnixd.md
@@ -223,6 +223,8 @@ If any verification or cutover command fails, leave Sinnixd stopped. Before the
 
 Both routes use the same UUID job ID, transient user service, cancellation, reconciliation, `job get/list/logs/result/wait`, and bounded artifact readers as declared operations. Their durable public record contains the principal, job kind, canonical project and checkout identity, redacted argv digest or prompt digest, and bounded artifact references. It never stores raw shell argv arguments after launch, prompt text, environment values, or credentials.
 
+Delivery is a precondition of `workspace.publish` and `workspace.land`, not a caller-fed completion route. Ordinary delivery reads the exact-head declared verification job through `job.result`. Packet delivery additionally names the Beads-bound attested-agent job with `--packet-job`. Both jobs receive the same immutable Beads binding at dispatch, which carries bead identity and write scope. The write scope covers exactly the packet job's immutable range from the start head recorded in the packet job's durable record to the final head the contract runner observes when it exits and seals the worker's structured report; it does not extend to arbitrary pre-existing branch content. The start head and final head are read exclusively from the two job records — the packet job's durable checkout head is the start, and the sealed envelope's `final_head` field (written by the runner at exit) is the end. Neither value comes from a field inside the Beads binding. Delivery requires the bindings to match, the later semantic verifier to succeed at that same final head, and snapshots the start-to-final Git range. It rejects dirty, divergent, stale, or out-of-scope work and repeats the complete precondition after push and after review inspection. The worker report can only tighten acceptance through bounded anti-vacuity, unresolved-work, delegation-visibility, deletion-evidence, and evidence-only fields. Deletion evidence must exactly equal the set of paths deleted in the packet range; overclaims (listing unrelated paths) are rejected as well as omissions. Git owns paths, commits, and heads; the project verifier owns semantic success; GitHub branch protection supplies independent review and required-check authority. Repositories without branch protection rules do not gain those approval requirements from AgentCTL. Beads closure consumes the returned completion artifact and bead references in its own owner; wiring that external closure consumer is not implemented by Sinnixd.
+
 Typed jobs accept no environment overlay. The daemon creates the `env -i` environment from the declared project environment and fixed `SINNIXD_*` identity fields. Immediately before execution, the contract runner verifies those fields, rechecks the exact registered project, canonical worktree root, common Git directory, porcelain worktree membership, and recorded HEAD. A changed, missing, symlinked, or spoofed identity fails closed. Agent handoff includes `--registered-project`, `--expected-git-common-dir`, and the canonical checkout path; nested scope creation remains disabled, so the native runner provides backend execution and native attestation while the shared transient user service remains the sole process, cgroup, timeout, and cancellation authority. Private launch inputs are mode 0600, removed before shell execution, and removed after agent handoff or every terminal lifecycle outcome, including confirmed launch failure. Native private logs are removed after handoff; only the bounded shared log and result artifacts remain addressable.
 
 Each record is stored under `$XDG_STATE_HOME/sinnixd` and contains safe operation identity, environment key names, and its bounded-read log artifact path. Record replacement fsyncs the containing directory, and newly created state directories are synchronized before they contain durable evidence. The `sinnixd-job-*.service` dynamic runtime surface and its record capture lane are declared with the daemon, rather than with any MCP frontend. Internal foreground argv is launch-only: the durable record has only a SHA-256 digest and constant display metadata, never raw argv or environment values. The systemd-launched capture helper drains output but writes at most 1 MiB per job; it creates its overflow marker with the first discarded byte, so a live log reader can see truncation before the producer exits. It also fsyncs a completion marker only after the captured process exits successfully and all bounded outputs are durable. It does not own a PID, process state, queue, task, workspace, or retry policy. A job ID deterministically derives its unit name. Every `systemd-run` and `systemctl` call has a short finite bound. `job.wait` caps each reconciliation call to its remaining deadline, so a stalled user manager cannot hold a wait or reserved control worker indefinitely. After a daemon restart, `get`, `list`, `wait`, and `cancel` reload the record and reconcile with the user manager. If `systemd-run` loses its reply but `show` finds the transient unit, `job start` returns the reconciled systemd state. If both the launch reply and its first reconciliation are unavailable, `job start` returns a durable nonterminal `launch-unknown` result with the stable job ID and unit. Later `get`, `wait`, and `cancel` use that same identity to reconcile it. A confirmed absent launch becomes terminal `launch-failed`. A confirmed missing unit after launch remains terminal `missing`; an unreachable or timed-out systemd observation is durable nonterminal `observation-unknown` until a later observation repairs it. Cancellation persists its intent before asking systemd to stop the service, then preserves an observed systemd success, timeout, or failure result. A `cancelled` result needs matching systemd signal evidence, or a durably recorded successful stop acknowledgement for the observed invocation when systemd has already garbage-collected the transient unit. If a stop times out and the unit later disappears, the job remains nonterminal `outcome-unknown` instead of treating the missing unit's default success fields as an exit result. A later authoritative systemd observation can repair that state. A typed result can prove semantic success after collection only when its content is valid and the capture completion marker proves the producer exited successfully; an empty, partial, malformed, or unmarked result is not completion evidence. A schema-v3 attested-agent record also carries forward its native completion only when systemd still reports an inactive loaded success, its durable lifecycle is `succeeded` with exit status zero, its bounded last-message artifact is valid, and no cancellation intent exists. Existing false terminal success or cancellation records without this evidence are reopened lazily by `get`, `list`, `wait`, or `cancel` and reconciled under the same rules. Systemd remains authoritative for the process, cgroup, timeout, terminal result, cancellation, and journal evidence.
diff --git a/pkgs/sinnix-agent-gateway/sinnix_agent_gateway/runtime.py b/pkgs/sinnix-agent-gateway/sinnix_agent_gateway/runtime.py
index 8fb00439..8069c2bb 100644
--- a/pkgs/sinnix-agent-gateway/sinnix_agent_gateway/runtime.py
+++ b/pkgs/sinnix-agent-gateway/sinnix_agent_gateway/runtime.py
@@ -875,6 +875,15 @@ def v2_run_for_bead(
             "request_id": request_id,
             "assignment_ref": parent_assignment_ref,
         }
+        metadata = bead.get("metadata")
+        encoded_scope = metadata.get("write_scope") if isinstance(metadata, Mapping) else None
+        if isinstance(encoded_scope, str):
+            try:
+                write_scope = json.loads(encoded_scope)
+            except json.JSONDecodeError:
+                write_scope = None
+            if isinstance(write_scope, list):
+                binding["write_scope"] = write_scope
         assigned_context = {
             "bead": bead,
             "project_ref": project_ref,
diff --git a/pkgs/sinnix-agent-gateway/test_execution_jobs.py b/pkgs/sinnix-agent-gateway/test_execution_jobs.py
index a2463b7b..7cd57b99 100644
--- a/pkgs/sinnix-agent-gateway/test_execution_jobs.py
+++ b/pkgs/sinnix-agent-gateway/test_execution_jobs.py
@@ -610,7 +610,7 @@ def test_agent_control_bead_scope_requires_matching_current_assignment(
     runtime, daemon = runtime_with_daemon(tmp_path, "agent-control")
     assignment_id = "3b0237a0-32a9-4f6b-a014-2a0ecfd2f75c"
     assignment_ref = f"sinnix://jobs/{assignment_id}"
-    bead = {"ref": "sinnix://projects/fixture/beads/fixture-1", "task_revision": "a" * 64, "etag": "b" * 64, "fields": {"title": "assigned"}}
+    bead = {"ref": "sinnix://projects/fixture/beads/fixture-1", "task_revision": "a" * 64, "etag": "b" * 64, "fields": {"title": "assigned"}, "metadata": {"write_scope": '["pkgs/sinnixd/"]'}}
     binding = {"bead_ref": bead["ref"], "project_ref": "sinnix://projects/fixture", "checkout_ref": "sinnix://projects/fixture/checkouts/default", "task_revision": "a" * 64, "task_etag": "b" * 64, "claim_ref": None, "claim_receipt": None, "request_id": "2e46daf5-e9b1-4c6e-b99d-bcd46631730b", "assignment_ref": None}
     daemon.responses["job.get"] = {"job_id": assignment_id, "principal": "agent-control", "state": {"phase": "running"}, "checkout": {"checkout_id": "default", "head": "c" * 40}, "contract": {"bead_binding": binding}, "artifacts": {"result": None}}
     daemon.responses["job.agent.start"] = {"job_id": "4a42f848-9057-4cef-9d27-80a022c0e16f", "state": {"phase": "running"}}
@@ -630,6 +630,7 @@ def test_agent_control_bead_scope_requires_matching_current_assignment(
     assert started["assignment_ref"] == assignment_ref
     assert daemon.calls[-1].principal == "agent-control"
     assert daemon.calls[-1].arguments["bead_binding"]["assignment_ref"] == assignment_ref
+    assert daemon.calls[-1].arguments["bead_binding"]["write_scope"] == ["pkgs/sinnixd/"]
     assert "private launch instruction" not in daemon.calls[-1].arguments["bead_binding"].values()
 
     foreign = {**binding, "bead_ref": "sinnix://projects/fixture/beads/fixture-2"}
diff --git a/pkgs/sinnixd/sinnixd/cli.py b/pkgs/sinnixd/sinnixd/cli.py
index a11a605f..5b3517e3 100644
--- a/pkgs/sinnixd/sinnixd/cli.py
+++ b/pkgs/sinnixd/sinnixd/cli.py
@@ -89,6 +89,7 @@ def parser() -> argparse.ArgumentParser:
     workspace_publish = workspace_subcommands.add_parser("publish")
     workspace_publish.add_argument("workspace_id")
     workspace_publish.add_argument("--job", required=True)
+    workspace_publish.add_argument("--packet-job")
     workspace_publish.add_argument("--title", required=True)
     workspace_publish.add_argument("--body", default="")
     workspace_review = workspace_subcommands.add_parser("review-status")
@@ -96,6 +97,7 @@ def parser() -> argparse.ArgumentParser:
     workspace_land = workspace_subcommands.add_parser("land")
     workspace_land.add_argument("workspace_id")
     workspace_land.add_argument("--job", required=True)
+    workspace_land.add_argument("--packet-job")
     workspace_finish = workspace_subcommands.add_parser("finish")
     workspace_finish.add_argument("workspace_id")
     workspace_finish_integrated = workspace_subcommands.add_parser("finish-integrated")
@@ -108,6 +110,7 @@ def parser() -> argparse.ArgumentParser:
     start.add_argument("operation")
     start.add_argument("--workspace")
     start.add_argument("--parameters-json", default="{}")
+    start.add_argument("--bead-binding-json")
     get = job_subcommands.add_parser("get")
     get.add_argument("job_id")
     status = job_subcommands.add_parser("status")
@@ -366,7 +369,13 @@ def main() -> int:
     elif arguments.command == "workspace" and arguments.workspace_command == "publish":
         request = _request(
             "workspace.publish", "git-workspaces",
-            {"workspace_id": arguments.workspace_id, "job_id": arguments.job, "title": arguments.title, "body": arguments.body},
+            {
+                "workspace_id": arguments.workspace_id,
+                "job_id": arguments.job,
+                "title": arguments.title,
+                "body": arguments.body,
+                **({"packet_job_id": arguments.packet_job} if arguments.packet_job else {}),
+            },
             "agent-control",
         )
     elif arguments.command == "workspace" and arguments.workspace_command == "review-status":
@@ -374,7 +383,11 @@ def main() -> int:
     elif arguments.command == "workspace" and arguments.workspace_command == "land":
         request = _request(
             "workspace.land", "git-workspaces",
-            {"workspace_id": arguments.workspace_id, "job_id": arguments.job}, "agent-control",
+            {
+                "workspace_id": arguments.workspace_id,
+                "job_id": arguments.job,
+                **({"packet_job_id": arguments.packet_job} if arguments.packet_job else {}),
+            }, "agent-control",
         )
     elif arguments.command == "workspace" and arguments.workspace_command == "finish-integrated":
         request = _request(
@@ -394,6 +407,14 @@ def main() -> int:
             parser().error(f"--parameters-json must be valid JSON: {error.msg}")
         if not isinstance(parameters, dict):
             parser().error("--parameters-json must be a JSON object")
+        binding = None
+        if arguments.bead_binding_json is not None:
+            try:
+                binding = json.loads(arguments.bead_binding_json)
+            except json.JSONDecodeError as error:
+                parser().error(f"--bead-binding-json must be valid JSON: {error.msg}")
+            if not isinstance(binding, dict):
+                parser().error("--bead-binding-json must be a JSON object")
         request = _request(
             "job.start",
             "systemd-jobs",
@@ -402,6 +423,7 @@ def main() -> int:
                 "operation": arguments.operation,
                 "workspace_id": arguments.workspace,
                 "parameters": parameters,
+                **({"bead_binding": binding} if binding is not None else {}),
             },
         )
     elif arguments.command == "job" and arguments.job_command in {"get", "status"}:
diff --git a/pkgs/sinnixd/sinnixd/contracts.py b/pkgs/sinnixd/sinnixd/contracts.py
index d31f92cc..6a8cdc9a 100644
--- a/pkgs/sinnixd/sinnixd/contracts.py
+++ b/pkgs/sinnixd/sinnixd/contracts.py
@@ -134,7 +134,7 @@ def start_agent(
         if not self.native_runner.is_file() or not os.access(self.native_runner, os.X_OK):
             raise ContractError("native agent runner is unavailable")
         checkout = self.projects.checkout(project_id, checkout_id)
-        binding = self._bead_binding(bead_binding, checkout)
+        binding = self.bead_binding(bead_binding, checkout)
         job_id = str(uuid4())
         prompt_path = self.inputs_root / f"{job_id}.prompt"
         public_contract = {
@@ -240,19 +240,35 @@ def _start(
         return response
 
     @staticmethod
-    def _bead_binding(
+    def bead_binding(
         value: Mapping[str, Any] | None, checkout: RegisteredCheckout
     ) -> dict[str, Any] | None:
-        """Validate public Beads provenance carried by an attested agent job."""
+        """Validate public Beads provenance frozen into a packet job contract."""
         if value is None:
             return None
         expected = {
             "bead_ref", "project_ref", "checkout_ref", "task_revision",
             "task_etag", "claim_ref", "claim_receipt", "request_id", "assignment_ref",
         }
-        if not isinstance(value, Mapping) or set(value) != expected:
+        allowed = expected | {"write_scope"}
+        if not isinstance(value, Mapping) or (set(value) != expected and set(value) != allowed):
             raise ContractError("agent bead binding is malformed")
         binding = dict(value)
+        scope = binding.get("write_scope")
+        if "write_scope" in binding and (
+            not isinstance(scope, list)
+            or not scope
+            or len(scope) > 128
+            or any(
+                not isinstance(path, str)
+                or not path
+                or len(path.encode()) > 1024
+                or path.startswith("/")
+                or ".." in Path(path).parts
+                for path in scope
+            )
+        ):
+            raise ContractError("agent Beads write scope is malformed")
         project_ref = f"sinnix://projects/{checkout.project_id}"
         checkout_ref = f"{project_ref}/checkouts/{checkout.checkout_id}"
         bead_prefix = f"{project_ref}/beads/"
@@ -293,7 +309,9 @@ def _bead_binding(
             UUID(str(binding["request_id"]))
         except (TypeError, ValueError, AttributeError) as error:
             raise ContractError("agent bead binding request_id is malformed") from error
-        return binding
+        # The caller retains its request object. Persist an independent JSON value so
+        # neither it nor a nested claim receipt can mutate a launched job's binding.
+        return json.loads(json.dumps(binding, sort_keys=True, separators=(",", ":")))
 
     def _environment(
         self, checkout: RegisteredCheckout, job_id: str, principal: str, timeout_seconds: int
diff --git a/pkgs/sinnixd/sinnixd/delivery.py b/pkgs/sinnixd/sinnixd/delivery.py
index 42be55da..b167cf41 100644
--- a/pkgs/sinnixd/sinnixd/delivery.py
+++ b/pkgs/sinnixd/sinnixd/delivery.py
@@ -4,6 +4,7 @@
 import subprocess
 import time
 from dataclasses import dataclass
+from pathlib import Path
 from typing import Any, Callable, Mapping, Sequence
 
 from .jobs import GenericJobs
@@ -25,14 +26,17 @@ class GitHubDelivery:
     jobs: GenericJobs
     run: Run = subprocess.run
 
-    def publish(self, workspace_id: str, job_id: str, title: str, body: str) -> dict[str, Any]:
-        workspace, project = self._verified_workspace(workspace_id, job_id)
+    def publish(
+        self, workspace_id: str, job_id: str, title: str, body: str, packet_job_id: str | None = None
+    ) -> dict[str, Any]:
+        workspace, project, receipt = self._verified_workspace(workspace_id, job_id, packet_job_id)
         if not title.strip() or len(title) > 256 or len(body.encode()) > 64_000:
             raise DeliveryError("review title or body exceeds its publication bounds")
         base = self._base_branch(project.workspace.default_base)
         path = workspace["path"]
         branch = workspace["branch"]
         self._command([*project.environment.command, "git", "-C", path, "push", "-u", "origin", branch], cwd=path)
+        workspace, project, receipt = self._verified_workspace(workspace_id, job_id, packet_job_id)
         existing = self.run(
             ["gh", "pr", "view", branch, "--json", "url"],
             cwd=path, capture_output=True, text=True, timeout=60, check=False,
@@ -47,7 +51,7 @@ def publish(self, workspace_id: str, job_id: str, title: str, body: str) -> dict
             ).stdout.strip()
             created = True
         review = self._review_after_push(workspace_id)
-        return {**review, "published": True, "created": created, "publication_output": publication_output}
+        return {**review, "published": True, "created": created, "publication_output": publication_output, "completion": receipt}
 
     def _review_after_push(self, workspace_id: str) -> dict[str, Any]:
         for attempt in range(10):
@@ -81,8 +85,8 @@ def review_status(self, workspace_id: str) -> dict[str, Any]:
             raise DeliveryError("GitHub review head does not match workspace HEAD")
         return {"workspace_id": workspace_id, "head": workspace["head"], "review": dict(review)}
 
-    def land(self, workspace_id: str, job_id: str) -> dict[str, Any]:
-        self._verified_workspace(workspace_id, job_id)
+    def land(self, workspace_id: str, job_id: str, packet_job_id: str | None = None) -> dict[str, Any]:
+        self._verified_workspace(workspace_id, job_id, packet_job_id)
         status = self.review_status(workspace_id)
         review = status["review"]
         if (
@@ -92,11 +96,12 @@ def land(self, workspace_id: str, job_id: str) -> dict[str, Any]:
             or not self._checks_pass(review["statusCheckRollup"])
         ):
             raise DeliveryError("review is not in a landable GitHub state")
+        _workspace, _project, receipt = self._verified_workspace(workspace_id, job_id, packet_job_id)
         self._command(["gh", "pr", "merge", str(review["number"]), "--squash"], cwd=self.workspaces.get(workspace_id)["path"])
         merged = self.review_status(workspace_id)
         if merged["review"]["state"] != "MERGED":
             raise DeliveryError("GitHub did not report the review merged")
-        return {**merged, "landed": True}
+        return {**merged, "landed": True, "completion": receipt}
 
     def finish(self, workspace_id: str) -> dict[str, Any]:
         status = self.review_status(workspace_id)
@@ -106,9 +111,11 @@ def finish(self, workspace_id: str) -> dict[str, Any]:
         self._delete_remote_branch(workspace["path"], workspace["branch"])
         return self.workspaces.finish_merged(workspace_id, status["head"])
 
-    def _verified_workspace(self, workspace_id: str, job_id: str) -> tuple[dict[str, Any], Any]:
+    def _verified_workspace(
+        self, workspace_id: str, job_id: str, packet_job_id: str | None = None
+    ) -> tuple[dict[str, Any], Any, dict[str, Any]]:
         workspace = self.workspaces.get(workspace_id)
-        if workspace["state"] != "available" or workspace["dirty"] or not workspace["identity_matches"]:
+        if workspace["state"] != "available" or not workspace["identity_matches"]:
             raise DeliveryError("publication requires an available clean identity-matching workspace")
         project = self.projects.get(workspace["project_id"])
         assert project.workspace is not None
@@ -119,11 +126,129 @@ def _verified_workspace(self, workspace_id: str, job_id: str) -> tuple[dict[str,
             job["state"].get("phase") != "succeeded"
             or checkout is None
             or checkout.get("checkout_id") != workspace["checkout_id"]
-            or checkout.get("head") != workspace["head"]
             or record.spec.operation not in project.workspace.verification_operations
         ):
             raise DeliveryError("workspace lacks successful declared verification at its exact HEAD")
-        return workspace, project
+        try:
+            self.jobs.result(job_id)
+            binding = self._binding(record, checkout, workspace) if packet_job_id is not None else None
+            packet = self._packet(packet_job_id, workspace, binding) if packet_job_id is not None else None
+            start_head = packet["start_head"] if packet is not None else checkout["head"]
+            scope = packet["scope"] if packet is not None else ()
+            snapshot = self.workspaces.delivery_snapshot(workspace_id, start_head, scope=scope)
+        except DeliveryError:
+            raise
+        except (ValueError, WorkspaceError) as error:
+            raise DeliveryError("workspace lacks an authoritative exact-head completion receipt") from error
+        if snapshot["head"] != checkout["head"] or not snapshot["descendant"] or snapshot["dirty"]:
+            raise DeliveryError("workspace lacks successful declared verification at its exact HEAD")
+        if packet is not None and (packet["final_head"] != snapshot["head"] or not snapshot["in_scope"]):
+            raise DeliveryError("packet delivery is outside its Beads-owned write scope")
+        if packet is not None:
+            self._validate_delivery_result(packet["delivery"], snapshot)
+        return workspace, project, {
+            "ref": packet["artifact_ref"] if packet is not None else f"sinnix://jobs/{job_id}",
+            "job_id": job_id,
+            "packet_job_id": packet_job_id,
+            "bead_ref": packet["bead_ref"] if packet is not None else None,
+            "workspace_id": workspace_id,
+            "head": snapshot["head"],
+            "verification_operation": record.spec.operation,
+        }
+
+    def _binding(
+        self, record: Any, checkout: Mapping[str, Any], workspace: Mapping[str, Any]
+    ) -> Mapping[str, Any]:
+        binding = record.spec.contract.get("bead_binding")
+        scope = binding.get("write_scope") if isinstance(binding, Mapping) else None
+        if (
+            not isinstance(binding, Mapping)
+            or not isinstance(scope, list)
+            or not scope
+            or len(scope) > 128
+            or any(
+                not isinstance(path, str)
+                or not path
+                or len(path.encode()) > 1024
+                or path.startswith("/")
+                or ".." in Path(path).parts
+                for path in scope
+            )
+            or checkout.get("checkout_id") != workspace.get("checkout_id")
+        ):
+            raise DeliveryError("declared verification lacks an authoritative Beads packet binding")
+        return binding
+
+    def _packet(
+        self, job_id: str, workspace: Mapping[str, Any], binding: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        job = self.jobs.get(job_id)
+        record = self.jobs.store.load(job_id)
+        checkout = record.spec.checkout
+        packet_binding = record.spec.contract.get("bead_binding")
+        result = self.jobs.result(job_id)
+        if (
+            job["state"].get("phase") != "succeeded"
+            or record.spec.kind != "attested-agent"
+            or not isinstance(checkout, Mapping)
+            or checkout.get("checkout_id") != workspace.get("checkout_id")
+            or packet_binding != binding
+            or result.get("kind") != "last-message"
+            or result.get("truncated") is not False
+            or not isinstance(result.get("content"), str)
+        ):
+            raise DeliveryError("packet job lacks an authoritative Beads-bound result")
+        try:
+            envelope = json.loads(result["content"])
+        except json.JSONDecodeError as error:
+            raise DeliveryError("packet job result is malformed") from error
+        if (
+            not isinstance(envelope, Mapping)
+            or set(envelope) != {"schema_version", "job_id", "start_head", "final_head", "delivery"}
+            or envelope.get("schema_version") != 1
+            or envelope.get("job_id") != job_id
+            or envelope.get("start_head") != checkout.get("head")
+            or not isinstance(envelope.get("final_head"), str)
+            or len(envelope["final_head"]) != 40
+            or any(value not in "0123456789abcdef" for value in envelope["final_head"])
+        ):
+            raise DeliveryError("packet job result identity is malformed")
+        artifact = result.get("artifact")
+        return {
+            "start_head": envelope["start_head"],
+            "final_head": envelope["final_head"],
+            "delivery": envelope["delivery"],
+            "scope": tuple(binding["write_scope"]),
+            "bead_ref": binding.get("bead_ref"),
+            "artifact_ref": artifact.get("ref") if isinstance(artifact, Mapping) else f"sinnix://jobs/{job_id}",
+        }
+
+    @staticmethod
+    def _validate_delivery_result(delivery: Any, snapshot: Mapping[str, Any]) -> None:
+        if not isinstance(delivery, Mapping) or set(delivery) != {"anti_vacuity", "unresolved_work", "delegation", "deletion_evidence", "evidence_only"}:
+            raise DeliveryError("project delivery result is malformed")
+        unresolved, delegation, deletions = delivery["unresolved_work"], delivery["delegation"], delivery["deletion_evidence"]
+        if delivery["anti_vacuity"] is not True or not isinstance(unresolved, list) or unresolved or not isinstance(deletions, list) or not isinstance(delivery["evidence_only"], bool) or not isinstance(delegation, Mapping) or set(delegation) != {"visibility", "pending"}:
+            raise DeliveryError("project delivery result is incomplete")
+        visibility, pending = delegation["visibility"], delegation["pending"]
+        if visibility not in {"supported", "unsupported"} or (visibility == "supported" and pending is not False) or (visibility == "unsupported" and pending is not None):
+            raise DeliveryError("project delivery delegation visibility is invalid")
+        changes = snapshot.get("changes")
+        if not isinstance(changes, list):
+            raise DeliveryError("workspace delivery snapshot is malformed")
+        deleted = {
+            path
+            for change in changes
+            if isinstance(change, Mapping) and str(change.get("status", ""))[:1] == "D"
+            for path in change.get("paths", [])
+            if isinstance(path, str)
+        }
+        if any(not isinstance(path, str) for path in deletions) or set(deletions) != deleted:
+            raise DeliveryError("project delivery result deletion evidence does not exactly match deleted paths")
+        if not changes and not delivery["evidence_only"]:
+            raise DeliveryError("no-change delivery lacks the evidence-only exception")
+        if changes and delivery["evidence_only"]:
+            raise DeliveryError("evidence-only delivery contains code changes")
 
     @staticmethod
     def _base_branch(default_base: str) -> str:
diff --git a/pkgs/sinnixd/sinnixd/jobs.py b/pkgs/sinnixd/sinnixd/jobs.py
index c49cef91..a6029133 100644
--- a/pkgs/sinnixd/sinnixd/jobs.py
+++ b/pkgs/sinnixd/sinnixd/jobs.py
@@ -1735,6 +1735,7 @@ def start_declared(
         parameters: Mapping[str, Any],
         checkout: RegisteredCheckout | None = None,
         principal: str = "operator",
+        contract: Mapping[str, Any] | None = None,
     ) -> dict[str, Any]:
         if principal not in {"agent-control", "operator"}:
             raise ValueError("declared operations require agent-control or operator principal")
@@ -1742,7 +1743,7 @@ def start_declared(
             raise ValueError("declared job checkout belongs to another project")
         with self._admission_lock:
             return self._start_declared_locked(
-                project, operation, correlation_id, principal, parameters, checkout, ()
+                project, operation, correlation_id, principal, parameters, checkout, (), contract or {}
             )
 
     def _start_declared_locked(
@@ -1754,6 +1755,7 @@ def _start_declared_locked(
         parameters: Mapping[str, Any],
         checkout: RegisteredCheckout | None,
         lineage: tuple[str, ...],
+        contract: Mapping[str, Any],
     ) -> dict[str, Any]:
         if operation.name in lineage:
             raise ValueError("declared operation dependency cycle")
@@ -1766,6 +1768,7 @@ def _start_declared_locked(
                 {},
                 checkout,
                 (*lineage, operation.name),
+                {},
             )
             for name in operation.dependencies
         )
@@ -1847,6 +1850,7 @@ def build_spec(lease: ServiceLease | None) -> GenericJobSpec:
                 principal=principal,
                 timeout_seconds=operation.timeout_seconds,
                 checkout=checkout.to_dict() if checkout is not None else None,
+                contract=dict(contract),
                 result_kind={"exit": "exit-status", "json": "json", "pytest": "pytest"}[operation.result],
                 pool=operation.pool, exclusive_keys=operation.exclusive_keys, dependency_job_ids=dependency_ids,
                 cache_key=cache_key, estimate_key=estimate_key, estimate_memory_bytes=estimate,
diff --git a/pkgs/sinnixd/sinnixd/runner.py b/pkgs/sinnixd/sinnixd/runner.py
index ab27bec5..284a13e0 100644
--- a/pkgs/sinnixd/sinnixd/runner.py
+++ b/pkgs/sinnixd/sinnixd/runner.py
@@ -7,7 +7,12 @@
 from pathlib import Path
 from typing import Any, Mapping, Sequence
 
-from .jobs import GenericJobStore, JobRecordError, MAX_RESULT_BYTES
+from .jobs import (
+    GenericJobStore,
+    JobRecordError,
+    MAX_RESULT_BYTES,
+    _open_preallocated_private_artifact,
+)
 from .limits import maximum_timeout_seconds, valid_timeout_seconds
 from .projects import ProjectConfigError, revalidate_registered_checkout
 
@@ -161,6 +166,9 @@ def _run_agent(
     ]
     try:
         completed = subprocess.run(command, cwd=checkout, check=False)
+        binding = value.get("bead_binding")
+        if isinstance(binding, Mapping) and "write_scope" in binding:
+            _seal_packet_result(value, checkout, result_path)
         if result_path.exists() and result_path.stat().st_size > MAX_RESULT_BYTES:
             result_path.write_bytes(result_path.read_bytes()[:MAX_RESULT_BYTES])
         return completed.returncode
@@ -168,6 +176,48 @@ def _run_agent(
         prompt_path.unlink(missing_ok=True)
 
 
+def _seal_packet_result(value: Mapping[str, Any], checkout: Path, result_path: Path) -> None:
+    """Bind a structured worker report to the runtime-observed terminal Git head."""
+    try:
+        if result_path.stat().st_size > MAX_RESULT_BYTES:
+            raise RunnerError("packet result exceeds the artifact limit")
+        raw = result_path.read_bytes()
+    except OSError as error:
+        raise RunnerError("packet result is unreadable") from error
+    try:
+        delivery = json.loads(raw)
+    except (UnicodeDecodeError, json.JSONDecodeError) as error:
+        raise RunnerError("packet result is not valid JSON") from error
+    observed = subprocess.run(
+        ["git", "-C", str(checkout), "rev-parse", "HEAD"],
+        capture_output=True,
+        text=True,
+        timeout=10,
+        check=False,
+    )
+    final_head = observed.stdout.strip()
+    if observed.returncode != 0 or len(final_head) != 40 or any(value not in "0123456789abcdef" for value in final_head):
+        raise RunnerError("packet final Git head is unavailable")
+    envelope = json.dumps(
+        {
+            "schema_version": 1,
+            "job_id": value["job_id"],
+            "start_head": value["checkout"]["head"],
+            "final_head": final_head,
+            "delivery": delivery,
+        },
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode()
+    if len(envelope) > MAX_RESULT_BYTES:
+        raise RunnerError("packet result exceeds the artifact limit")
+    with _open_preallocated_private_artifact(result_path) as result_file:
+        os.ftruncate(result_file.fileno(), 0)
+        result_file.write(envelope)
+        result_file.flush()
+        os.fsync(result_file.fileno())
+
+
 def main(arguments: Sequence[str] | None = None) -> int:
     parser = argparse.ArgumentParser(prog="sinnixd-contract-runner")
     parser.add_argument("--input", type=Path)
diff --git a/pkgs/sinnixd/sinnixd/service.py b/pkgs/sinnixd/sinnixd/service.py
index 0c32aca7..34d795f5 100644
--- a/pkgs/sinnixd/sinnixd/service.py
+++ b/pkgs/sinnixd/sinnixd/service.py
@@ -298,24 +298,35 @@ def _dispatch(
         if operation == "workspace.publish":
             if principal not in {"agent-control", "operator"}:
                 raise ValueError("workspace publication requires agent-control or operator principal")
-            if set(arguments) != {"workspace_id", "job_id", "title", "body"}:
+            if set(arguments) - {"workspace_id", "job_id", "packet_job_id", "title", "body"} or not {
+                "workspace_id", "job_id", "title", "body"
+            } <= set(arguments):
                 raise ValueError("workspace.publish requires workspace_id, job_id, title, and body")
             assert self.delivery is not None
-            return self.delivery.publish(
+            publish_arguments = (
                 self._job_argument(arguments, "workspace_id"),
                 self._job_argument(arguments, "job_id"),
                 self._job_argument(arguments, "title"),
                 arguments.get("body") if isinstance(arguments.get("body"), str) else "",
             )
+            packet_job_id = arguments.get("packet_job_id")
+            return self.delivery.publish(
+                *publish_arguments,
+                **({"packet_job_id": packet_job_id} if isinstance(packet_job_id, str) else {}),
+            )
         if operation == "workspace.review-status":
             assert self.delivery is not None
             return self.delivery.review_status(self._single_workspace_id(arguments, "workspace.review-status"))
         if operation == "workspace.land":
-            if principal not in {"agent-control", "operator"} or set(arguments) != {"workspace_id", "job_id"}:
+            if principal not in {"agent-control", "operator"} or set(arguments) - {
+                "workspace_id", "job_id", "packet_job_id"
+            } or not {"workspace_id", "job_id"} <= set(arguments):
                 raise ValueError("workspace.land requires agent-control or operator plus workspace_id and job_id")
             assert self.delivery is not None
+            packet_job_id = arguments.get("packet_job_id")
             return self.delivery.land(
-                self._job_argument(arguments, "workspace_id"), self._job_argument(arguments, "job_id")
+                self._job_argument(arguments, "workspace_id"), self._job_argument(arguments, "job_id"),
+                **({"packet_job_id": packet_job_id} if isinstance(packet_job_id, str) else {}),
             )
         if operation == "workspace.finish":
             if principal not in {"agent-control", "operator"}:
@@ -337,8 +348,8 @@ def _dispatch(
                 )
             project_id = self._job_argument(arguments, "project_id")
             operation_name = self._job_argument(arguments, "operation")
-            if set(arguments) - {"project_id", "operation", "workspace_id", "parameters"}:
-                raise ValueError("job.start accepts project_id, operation, optional workspace_id, and optional parameters")
+            if set(arguments) - {"project_id", "operation", "workspace_id", "parameters", "bead_binding"}:
+                raise ValueError("job.start accepts project_id, operation, optional workspace_id, optional parameters, and optional bead_binding")
             parameters = arguments.get("parameters", {})
             if not isinstance(parameters, Mapping):
                 raise ValueError("job.start parameters must be an object")
@@ -352,6 +363,14 @@ def _dispatch(
                 if workspace_id is not None
                 else self.projects.checkout(project_id, "default")
             )
+            binding = arguments.get("bead_binding")
+            if binding is not None and operation_name not in project.workspace.verification_operations:
+                raise ValueError("a Beads packet binding requires a declared verification operation")
+            packet_contract = (
+                {"bead_binding": self.job_contracts.bead_binding(binding, checkout)}
+                if binding is not None
+                else {}
+            )
             return self._cleanup_terminal(self.jobs.start_declared(
                 project=project,
                 operation=project.operation(operation_name),
@@ -359,6 +378,7 @@ def _dispatch(
                 principal=principal,
                 parameters=parameters,
                 checkout=checkout,
+                contract=packet_contract,
             ))
         if operation == "job.shell.start":
             required = {"project_id", "checkout_id", "argv", "cwd", "timeout_seconds", "result"}
diff --git a/pkgs/sinnixd/sinnixd/workspaces.py b/pkgs/sinnixd/sinnixd/workspaces.py
index cbfd6cf6..1d7466e8 100644
--- a/pkgs/sinnixd/sinnixd/workspaces.py
+++ b/pkgs/sinnixd/sinnixd/workspaces.py
@@ -13,7 +13,7 @@
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from pathlib import Path
-from typing import Any, Mapping
+from typing import Any, Mapping, Sequence
 from uuid import uuid4
 
 from sinnix_lib.atomic_json import modify_json, read_json, write_json_atomic
@@ -337,6 +337,22 @@ def get(self, workspace_id: str) -> dict[str, Any]:
         record = self._record(workspace_id)
         return self._status(record)
 
+    def delivery_snapshot(self, workspace_id: str, start_head: str, *, scope: Sequence[str] = ()) -> dict[str, Any]:
+        """Read one exact-head Git fact set for a delivery precondition."""
+        record = self._record(workspace_id)
+        checkout, _project = self._available(record)
+        before = self._git(checkout.path, "rev-parse", "HEAD").stdout.strip()
+        if before != checkout.head:
+            raise WorkspaceError("workspace HEAD changed during delivery snapshot")
+        descendant = self._git(checkout.path, "merge-base", "--is-ancestor", start_head, before, check=False).returncode == 0
+        changes = self._name_status(checkout.path, start_head, before)
+        dirty = self._porcelain_status(checkout.path)
+        after = self._git(checkout.path, "rev-parse", "HEAD").stdout.strip()
+        if after != before:
+            raise WorkspaceError("workspace HEAD changed during delivery snapshot")
+        paths = tuple(path for change in changes for path in change["paths"])
+        return {"workspace_id": workspace_id, "checkout_id": checkout.checkout_id, "start_head": start_head, "head": before, "descendant": descendant, "dirty": bool(dirty), "status": dirty, "changes": changes, "in_scope": all(self._scope_contains(path, scope) for path in paths) if scope else True}
+
     def checkout(self, workspace_id: str) -> RegisteredCheckout:
         record = self._record(workspace_id)
         checkout, _project = self._available(record)
@@ -776,6 +792,67 @@ def _status(self, record: WorkspaceRecord) -> dict[str, Any]:
             row.update({"state": "missing", "checkout_id": None, "head": None, "current_branch": None, "dirty": None, "identity_matches": False})
         return row
 
+    @classmethod
+    def _porcelain_status(cls, path: Path) -> list[dict[str, Any]]:
+        raw = cls._git_bytes(path, "status", "--porcelain=v1", "-z", "--untracked-files=all")
+        records = [item for item in raw.split(b"\0") if item]
+        result: list[dict[str, Any]] = []
+        index = 0
+        while index < len(records):
+            entry = records[index]
+            if len(entry) < 4 or entry[2:3] != b" ":
+                raise WorkspaceError("Git status porcelain is malformed")
+            status = entry[:2].decode("ascii", errors="strict")
+            paths = [cls._decode_git_path(entry[3:])]
+            index += 1
+            if "R" in status or "C" in status:
+                if index == len(records):
+                    raise WorkspaceError("Git status rename porcelain is malformed")
+                paths.append(cls._decode_git_path(records[index]))
+                index += 1
+            result.append({"status": status, "paths": paths})
+        return result
+
+    @classmethod
+    def _name_status(cls, path: Path, start_head: str, head: str) -> list[dict[str, Any]]:
+        raw = cls._git_bytes(path, "diff", "--name-status", "-z", "--find-renames", start_head, head, "--")
+        records = [item for item in raw.split(b"\0") if item]
+        result: list[dict[str, Any]] = []
+        index = 0
+        while index < len(records):
+            status = records[index].decode("ascii", errors="strict")
+            if not status or status[0] not in "ACDMRTUXB":
+                raise WorkspaceError("Git diff name-status porcelain is malformed")
+            index += 1
+            count = 2 if status[0] in {"R", "C"} else 1
+            if len(records) - index < count:
+                raise WorkspaceError("Git diff rename porcelain is malformed")
+            result.append({"status": status, "paths": [cls._decode_git_path(item) for item in records[index:index + count]]})
+            index += count
+        return result
+
+    @staticmethod
+    def _decode_git_path(value: bytes) -> str:
+        try:
+            path = value.decode()
+        except UnicodeDecodeError as error:
+            raise WorkspaceError("Git path is not UTF-8") from error
+        if not path or Path(path).is_absolute() or ".." in Path(path).parts:
+            raise WorkspaceError("Git path is unsafe")
+        return path
+
+    @staticmethod
+    def _scope_contains(path: str, scope: Sequence[str]) -> bool:
+        for entry in scope:
+            if not isinstance(entry, str) or not entry or entry.startswith("/") or ".." in Path(entry).parts:
+                raise WorkspaceError("delivery scope is unsafe")
+            if entry.endswith("/"):
+                if path.startswith(entry):
+                    return True
+            elif path == entry:
+                return True
+        return False
+
     def _record(self, workspace_id: str) -> WorkspaceRecord:
         for record in self.store.records():
             if record.workspace_id == workspace_id:
diff --git a/pkgs/sinnixd/test_service.py b/pkgs/sinnixd/test_service.py
index 694e9b06..20121d3d 100644
--- a/pkgs/sinnixd/test_service.py
+++ b/pkgs/sinnixd/test_service.py
@@ -59,7 +59,14 @@
 from sinnixd.limits import MAX_DECLARED_OPERATION_TIMEOUT_SECONDS
 from sinnixd.owner_adapters import DeclaredOwnerAdapters, OwnerAdapterError
 from sinnixd.projects import ProjectCatalog, ProjectConfigError, RegisteredCheckout, parse_worktree_records
-from sinnixd.runner import RunnerError, _exec_shell, _require_environment, _revalidate_checkout, _run_declared
+from sinnixd.runner import (
+    RunnerError,
+    _exec_shell,
+    _require_environment,
+    _revalidate_checkout,
+    _run_declared,
+    _seal_packet_result,
+)
 from sinnixd.service import SinnixdService
 from sinnixd.tasks import (
     BeadsCommandBoundary,
@@ -4463,6 +4470,424 @@ def test_declared_job_binds_workspace_and_exact_head(tmp_path: Path) -> None:
     assert record.spec.checkout["head"] == workspace["head"]
 
 
+def test_forged_packet_completion_arguments_have_no_service_route(tmp_path: Path) -> None:
+    write_adapter(tmp_path)
+    initialize_git_checkout(tmp_path)
+    jobs = generic_jobs(tmp_path)
+    service = SinnixdService(ProjectCatalog([tmp_path]), jobs=jobs)
+    workspace = service.workspaces.create(
+        project_id="fixture", name="packet-lane", branch="feature/packet-lane", base="HEAD"
+    )
+    started = service.dispatch(
+        request(
+            "job.start",
+            "systemd-jobs",
+            {"project_id": "fixture", "operation": "check", "workspace_id": workspace["workspace_id"]},
+        )
+    )
+    assert started.ok and started.payload is not None
+    job_id = started.payload.inline["job_id"]
+
+    response = service.dispatch(
+        request(
+            "job.packet-completion",
+            "systemd-jobs",
+            {
+                "job_id": job_id,
+                "workspace_id": workspace["workspace_id"],
+                "contract": {
+                    "job_id": job_id,
+                    "workspace_id": workspace["workspace_id"],
+                    "write_scope": ["src/"],
+                    "required_verification_refs": [],
+                },
+                "worker_result": None,
+                "verification_receipts": [],
+                "delegation": {"visibility": "unsupported", "pending": None},
+                "evidence_receipts": [],
+                "review": None,
+            },
+        )
+    )
+
+    assert response.error is not None
+    assert response.error.code.value == "INVALID_ARGUMENT"
+
+
+def test_delivery_snapshot_is_nul_safe_and_exact_file_scope_does_not_include_descendants(tmp_path: Path) -> None:
+    write_adapter(tmp_path)
+    service = SinnixdService(ProjectCatalog([tmp_path]), jobs=generic_jobs(tmp_path))
+    workspace = service.workspaces.create(project_id="fixture", name="snapshot-lane", branch="feature/snapshot", base="HEAD")
+    path = Path(workspace["path"])
+    (path / "dir").mkdir()
+    (path / "dir" / "exact").write_text("old\n")
+    (path / "dir" / "delete\nfile").write_text("delete\n")
+    subprocess.run(["git", "-C", str(path), "add", "."], check=True)
+    subprocess.run(["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "seed"], check=True)
+    start = service.workspaces.get(workspace["workspace_id"])["head"]
+    subprocess.run(["git", "-C", str(path), "mv", "dir/exact", "dir/renamed\nfile"], check=True)
+    (path / "dir" / "delete\nfile").unlink()
+    (path / "dir" / "exact.child").write_text("outside exact-file scope\n")
+    subprocess.run(["git", "-C", str(path), "add", "-A"], check=True)
+    subprocess.run(["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "paths"], check=True)
+    snapshot = service.workspaces.delivery_snapshot(workspace["workspace_id"], start, scope=("dir/exact",))
+    assert not snapshot["in_scope"]
+    assert {change["status"][0] for change in snapshot["changes"]} >= {"D", "R", "A"}
+    assert any("\n" in item for change in snapshot["changes"] for item in change["paths"])
+    assert service.workspaces.delivery_snapshot(workspace["workspace_id"], start, scope=("dir/",))["in_scope"]
+
+
+def test_beads_bound_packet_and_exact_head_verifier_compose_into_delivery(tmp_path: Path) -> None:
+    """The accepting path joins two authoritative jobs; neither can substitute for the other."""
+    write_adapter(tmp_path)
+    initialize_git_checkout(tmp_path)
+    native = tmp_path / "native-runner"
+    native_runner(native)
+    systemd = FakeSystemdJobs()
+    jobs = generic_jobs(tmp_path, systemd)
+    service = SinnixdService(ProjectCatalog([tmp_path]), jobs=jobs, native_runner=native)
+    workspace = service.workspaces.create(
+        project_id="fixture", name="packet-delivery", branch="feature/packet-delivery", base="HEAD"
+    )
+    checkout_id = workspace["checkout_id"]
+    binding = {
+        "bead_ref": "sinnix://projects/fixture/beads/fixture-1",
+        "project_ref": "sinnix://projects/fixture",
+        "checkout_ref": f"sinnix://projects/fixture/checkouts/{checkout_id}",
+        "task_revision": "a" * 64,
+        "task_etag": "b" * 64,
+        "claim_ref": f"sinnix://projects/fixture/beads/fixture-1/claims/{'c' * 64}",
+        "claim_receipt": {
+            "ref": f"sinnix://projects/fixture/beads/fixture-1/claims/{'c' * 64}",
+            "owner_route": "beads.cli",
+        },
+        "request_id": "2e46daf5-e9b1-4c6e-b99d-bcd46631730b",
+        "assignment_ref": None,
+        "write_scope": ["delivery.txt", "obsolete.txt"],
+    }
+    path = Path(workspace["path"])
+    (path / "obsolete.txt").write_text("remove me\n")
+    subprocess.run(["git", "-C", str(path), "add", "obsolete.txt"], check=True)
+    subprocess.run(
+        ["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "seed deletion"],
+        check=True,
+    )
+    packet = service.dispatch(
+        request(
+            "job.agent.start",
+            "systemd-jobs",
+            {
+                "project_id": "fixture", "checkout_id": checkout_id, "prompt": "return structured delivery",
+                "backend": "codex", "model": "fixture", "effort": "high",
+                "credential_profile": "subscription", "timeout_seconds": 60,
+                "result": "last-message", "bead_binding": binding,
+            },
+            "agent-control",
+        )
+    )
+    assert packet.ok and packet.payload is not None
+    packet_id = packet.payload.inline["job_id"]
+    packet_record = jobs.store.load(packet_id)
+    start_head = packet_record.spec.checkout["head"]
+    (path / "delivery.txt").write_text("delivered\n")
+    (path / "obsolete.txt").unlink()
+    subprocess.run(["git", "-C", str(path), "add", "-A"], check=True)
+    subprocess.run(
+        ["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "delivery"],
+        check=True,
+    )
+    final_head = subprocess.run(
+        ["git", "-C", str(path), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    worker_delivery = {
+        "anti_vacuity": True,
+        "unresolved_work": [],
+        "delegation": {"visibility": "unsupported", "pending": None},
+        "deletion_evidence": ["obsolete.txt"],
+        "evidence_only": False,
+    }
+    assert packet_record.result_path is not None
+    packet_record.result_path.write_text(json.dumps({
+        "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+        "final_head": final_head, "delivery": worker_delivery,
+    }))
+    jobs_module._write_private_marker(jobs_module._completion_marker_path(packet_record.log_path))
+    systemd.properties = {
+        "LoadState": "loaded", "ActiveState": "inactive", "Result": "success",
+        "ExecMainStatus": "0", "InvocationID": "fixture-invocation",
+    }
+    assert jobs.get(packet_id)["state"]["phase"] == "succeeded"
+    verifier = service.dispatch(request(
+        "job.start", "systemd-jobs",
+        {
+            "project_id": "fixture", "operation": "check", "workspace_id": workspace["workspace_id"],
+            "bead_binding": binding,
+        },
+    ))
+    assert verifier.ok and verifier.payload is not None
+    verifier_id = verifier.payload.inline["job_id"]
+    assert jobs.get(verifier_id)["state"]["phase"] == "succeeded"
+
+    delivery = GitHubDelivery(service.projects, service.workspaces, jobs)
+    _workspace, _project, receipt = delivery._verified_workspace(
+        workspace["workspace_id"], verifier_id, packet_id
+    )
+    assert receipt["bead_ref"] == binding["bead_ref"]
+    assert receipt["head"] == final_head
+
+    (path / "dirty.txt").write_text("uncommitted\n")
+    with pytest.raises(DeliveryError, match="exact HEAD"):
+        delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+    (path / "dirty.txt").unlink()
+
+    bad_binding = {**binding, "write_scope": ["other.txt"]}
+    jobs.store.save(replace(
+        packet_record,
+        spec=replace(packet_record.spec, contract={**packet_record.spec.contract, "bead_binding": bad_binding}),
+    ))
+    verifier_record = jobs.store.load(verifier_id)
+    jobs.store.save(replace(
+        verifier_record,
+        spec=replace(verifier_record.spec, contract={**verifier_record.spec.contract, "bead_binding": bad_binding}),
+    ))
+    with pytest.raises(DeliveryError, match="write scope"):
+        delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+    jobs.store.save(packet_record)
+    jobs.store.save(verifier_record)
+
+    for evidence in ([], ["unrelated.txt"]):
+        packet_record.result_path.write_text(json.dumps({
+            "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+            "final_head": final_head,
+            "delivery": {**worker_delivery, "deletion_evidence": evidence},
+        }))
+        with pytest.raises(DeliveryError, match="deletion evidence"):
+            delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+
+    packet_record.result_path.write_text(json.dumps({
+        "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+        "final_head": final_head,
+        "delivery": {**worker_delivery, "unresolved_work": ["still running"]},
+    }))
+    with pytest.raises(DeliveryError, match="incomplete"):
+        delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+    packet_record.result_path.write_text(json.dumps({
+        "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+        "final_head": final_head, "delivery": worker_delivery,
+    }))
+
+    packet_record.result_path.write_text(json.dumps({
+        "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+        "final_head": final_head, "delivery": None,
+    }))
+    with pytest.raises(DeliveryError, match="malformed"):
+        delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+    packet_record.result_path.write_text(json.dumps({
+        "schema_version": 1, "job_id": packet_id, "start_head": start_head,
+        "final_head": final_head, "delivery": worker_delivery,
+    }))
+
+    for scope in (["../outside"], [f"entry-{index}" for index in range(129)]):
+        rejected = service.dispatch(request(
+            "job.start", "systemd-jobs",
+            {
+                "project_id": "fixture", "operation": "check", "workspace_id": workspace["workspace_id"],
+                "bead_binding": {**binding, "write_scope": scope},
+            },
+        ))
+        assert not rejected.ok
+
+    (path / "later.txt").write_text("post-terminal\n")
+    subprocess.run(["git", "-C", str(path), "add", "later.txt"], check=True)
+    subprocess.run(
+        ["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "later"],
+        check=True,
+    )
+    with pytest.raises(DeliveryError, match="exact HEAD"):
+        delivery._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+
+
+def test_packet_runner_seals_worker_report_to_runtime_observed_head(tmp_path: Path) -> None:
+    initialize_git_checkout(tmp_path)
+    start_head = subprocess.run(
+        ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    (tmp_path / "change.txt").write_text("change\n")
+    subprocess.run(["git", "-C", str(tmp_path), "add", "change.txt"], check=True)
+    subprocess.run(
+        ["git", "-C", str(tmp_path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--quiet", "-m", "change"],
+        check=True,
+    )
+    final_head = subprocess.run(
+        ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    result_root = tmp_path / "private-results"
+    result_root.mkdir(mode=0o700)
+    result_path = result_root / "packet.result"
+    result_path.touch(mode=0o600)
+    delivery = {
+        "anti_vacuity": True, "unresolved_work": [],
+        "delegation": {"visibility": "unsupported", "pending": None},
+        "deletion_evidence": [], "evidence_only": False,
+    }
+    result_path.write_text(json.dumps(delivery))
+
+    _seal_packet_result(
+        {"job_id": "packet-job", "checkout": {"head": start_head}}, tmp_path, result_path
+    )
+
+    assert json.loads(result_path.read_text()) == {
+        "schema_version": 1,
+        "job_id": "packet-job",
+        "start_head": start_head,
+        "final_head": final_head,
+        "delivery": delivery,
+    }
+
+
+def test_packet_runner_rejects_invalid_utf8_as_typed_json_failure(tmp_path: Path) -> None:
+    initialize_git_checkout(tmp_path)
+    start_head = subprocess.run(
+        ["git", "-C", str(tmp_path), "rev-parse", "HEAD"], capture_output=True, text=True, check=True
+    ).stdout.strip()
+    result_path = tmp_path / "packet.result"
+    result_path.write_bytes(b"\xff")
+
+    with pytest.raises(RunnerError, match="packet result is not valid JSON") as caught:
+        _seal_packet_result(
+            {"job_id": "packet-job", "checkout": {"head": start_head}}, tmp_path, result_path
+        )
+
+    assert isinstance(caught.value.__cause__, UnicodeDecodeError)
+
+
+def test_seal_output_composes_through_exact_head_into_delivery_validation(tmp_path: Path) -> None:
+    """Composed: real runner seal output flows through exact-head evidence into delivery acceptance and tamper rejection."""
+    write_adapter(tmp_path)
+    initialize_git_checkout(tmp_path)
+    native = tmp_path / "native-runner"
+    native_runner(native)
+    systemd = FakeSystemdJobs()
+    jobs = generic_jobs(tmp_path, systemd)
+    service = SinnixdService(ProjectCatalog([tmp_path]), jobs=jobs, native_runner=native)
+    workspace = service.workspaces.create(
+        project_id="fixture", name="seal-compose", branch="feature/seal-compose", base="HEAD"
+    )
+    checkout_id = workspace["checkout_id"]
+    path = Path(workspace["path"])
+
+    # Seed a file that will be deleted during the packet range.
+    (path / "seed.txt").write_text("to be removed\n")
+    subprocess.run(["git", "-C", str(path), "add", "seed.txt"], check=True)
+    subprocess.run(
+        ["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test",
+         "commit", "--quiet", "-m", "seed deletion target"],
+        check=True,
+    )
+
+    binding = {
+        "bead_ref": "sinnix://projects/fixture/beads/seal-test-1",
+        "project_ref": "sinnix://projects/fixture",
+        "checkout_ref": f"sinnix://projects/fixture/checkouts/{checkout_id}",
+        "task_revision": "a" * 64,
+        "task_etag": "b" * 64,
+        "claim_ref": f"sinnix://projects/fixture/beads/seal-test-1/claims/{'c' * 64}",
+        "claim_receipt": {
+            "ref": f"sinnix://projects/fixture/beads/seal-test-1/claims/{'c' * 64}",
+            "owner_route": "beads.cli",
+        },
+        "request_id": "9f1a2b3c-0000-4d5e-8f6a-7b8c9d0e1f2a",
+        "assignment_ref": None,
+        "write_scope": ["added.txt", "seed.txt"],
+    }
+
+    packet_response = service.dispatch(request(
+        "job.agent.start", "systemd-jobs",
+        {
+            "project_id": "fixture", "checkout_id": checkout_id,
+            "prompt": "return structured delivery for seal composition test",
+            "backend": "codex", "model": "fixture", "effort": "high",
+            "credential_profile": "subscription", "timeout_seconds": 60,
+            "result": "last-message", "bead_binding": binding,
+        },
+        "agent-control",
+    ))
+    assert packet_response.ok and packet_response.payload is not None
+    packet_id = packet_response.payload.inline["job_id"]
+    packet_record = jobs.store.load(packet_id)
+    start_head = packet_record.spec.checkout["head"]
+
+    # Produce the packet range: one add, one delete.
+    (path / "added.txt").write_text("new content\n")
+    (path / "seed.txt").unlink()
+    subprocess.run(["git", "-C", str(path), "add", "-A"], check=True)
+    subprocess.run(
+        ["git", "-C", str(path), "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test",
+         "commit", "--quiet", "-m", "packet range: add+delete"],
+        check=True,
+    )
+
+    # Use the real runner seal on a valid worker delivery report.
+    assert packet_record.result_path is not None
+    packet_record.result_path.parent.mkdir(parents=True, exist_ok=True)
+    packet_record.result_path.touch(mode=0o600)
+    worker_delivery = {
+        "anti_vacuity": True,
+        "unresolved_work": [],
+        "delegation": {"visibility": "unsupported", "pending": None},
+        "deletion_evidence": ["seed.txt"],
+        "evidence_only": False,
+    }
+    packet_record.result_path.write_text(json.dumps(worker_delivery))
+    _seal_packet_result(
+        {"job_id": packet_id, "checkout": {"head": start_head}}, path, packet_record.result_path
+    )
+    sealed = json.loads(packet_record.result_path.read_text())
+    final_head = sealed["final_head"]
+
+    # Worker result was sealed by the real runner; mark the job succeeded.
+    jobs_module._write_private_marker(jobs_module._completion_marker_path(packet_record.log_path))
+    systemd.properties = {
+        "LoadState": "loaded", "ActiveState": "inactive", "Result": "success",
+        "ExecMainStatus": "0", "InvocationID": "seal-compose-invocation",
+    }
+    assert jobs.get(packet_id)["state"]["phase"] == "succeeded"
+
+    # Verifier job runs at final_head with the same binding.
+    verifier_response = service.dispatch(request(
+        "job.start", "systemd-jobs",
+        {
+            "project_id": "fixture", "operation": "check",
+            "workspace_id": workspace["workspace_id"], "bead_binding": binding,
+        },
+    ))
+    assert verifier_response.ok and verifier_response.payload is not None
+    verifier_id = verifier_response.payload.inline["job_id"]
+    assert jobs.get(verifier_id)["state"]["phase"] == "succeeded"
+
+    delivery_gate = GitHubDelivery(service.projects, service.workspaces, jobs)
+
+    # Accepting path: real seal output + correct deletion evidence + verifier at final_head.
+    _workspace, _project, receipt = delivery_gate._verified_workspace(
+        workspace["workspace_id"], verifier_id, packet_id
+    )
+    assert receipt["head"] == final_head
+    assert receipt["bead_ref"] == binding["bead_ref"]
+
+    # Tamper: mutate the sealed envelope's final_head to a synthetic value.
+    tampered = {**sealed, "final_head": "b" * 40}
+    packet_record.result_path.write_text(json.dumps(tampered))
+    with pytest.raises(DeliveryError):
+        delivery_gate._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+
+    # Restore and verify deletion overclaim is now rejected.
+    packet_record.result_path.write_text(json.dumps(sealed))
+    overclaim = {**worker_delivery, "deletion_evidence": ["seed.txt", "unrelated.txt"]}
+    packet_record.result_path.write_text(json.dumps({**sealed, "delivery": overclaim}))
+    with pytest.raises(DeliveryError, match="deletion evidence"):
+        delivery_gate._verified_workspace(workspace["workspace_id"], verifier_id, packet_id)
+
+
 def test_admission_revalidates_queued_declared_workspace_before_systemd_launch(tmp_path: Path) -> None:
     """A queued declared service whose checkout HEAD moved must terminalize before it reaches systemd."""
     write_adapter(tmp_path)