deploy(hf): sync szl-holdings/a11oy@5c711ffb13b8ce4e0517f03c73700eaf8bfea3bb derived COPY set

Reusable Dockerfile-COPY-derived deploy from szl-holdings/a11oy 5c711ffb13b8ce4e0517f03c73700eaf8bfea3bb.
Files: 1179 Pruned: 0
Derived from Dockerfile COPY sources (NO hand-maintained allowlist).

Signed-off-by: SZL Holdings <noreply@szlholdings.ai>
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

gdw_proofs.py

@@ -1,15 +1,30 @@
 """Structured theorem-input export for asynchronous Lean checking."""
 
+import errno
 import hashlib
 import json
 import os
+import stat
 import tempfile
 import threading
+import time
 from pathlib import Path
 from typing import Any, Dict
 
 
 _ARTIFACT_QUOTA_LOCK = threading.RLock()
+_TRANSIENT_LINK_ERRNOS = {
+    errno.ENOTSUP,
+    getattr(errno, "EOPNOTSUPP", errno.ENOTSUP),
+}
+_UNSUPPORTED_DIRECTORY_FSYNC_ERRNOS = {
+    errno.ENOSYS,
+    errno.EINVAL,
+    errno.ENOTSUP,
+    getattr(errno, "EOPNOTSUPP", errno.ENOTSUP),
+}
+_LINK_MAX_ATTEMPTS = 61
+_LINK_RETRY_SECONDS = 0.5
 
 
 def canonical_json(value: Any) -> str:
@@ -20,6 +35,46 @@ def sha256_json(value: Any) -> str:
     return hashlib.sha256(canonical_json(value).encode("utf-8")).hexdigest()
 
 
+def _read_existing_regular(path: Path) -> bytes:
+    flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0)
+    try:
+        descriptor = os.open(path, flags)
+    except OSError as exc:
+        if exc.errno == getattr(errno, "ELOOP", None):
+            raise FileExistsError(
+                "existing GDW artifact is not a regular file"
+            ) from None
+        raise
+    try:
+        if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+            raise FileExistsError(
+                "existing GDW artifact is not a regular file"
+            )
+        with os.fdopen(descriptor, "rb") as stream:
+            descriptor = -1
+            return stream.read()
+    finally:
+        if descriptor >= 0:
+            os.close(descriptor)
+
+
+def _fsync_directory(path: Path) -> None:
+    if os.name != "posix":
+        return
+    descriptor = os.open(
+        path,
+        os.O_RDONLY | getattr(os, "O_DIRECTORY", 0),
+    )
+    try:
+        try:
+            os.fsync(descriptor)
+        except OSError as exc:
+            if exc.errno not in _UNSUPPORTED_DIRECTORY_FSYNC_ERRNOS:
+                raise
+    finally:
+        os.close(descriptor)
+
+
 def build_proof_payload(
     proposal_id: str,
     request_id: str,
@@ -89,11 +144,12 @@ def _export_json_artifact_unlocked(
     destination = owner_root / filename
     encoded = (json.dumps(payload, indent=2, sort_keys=True) + "\n").encode("utf-8")
     expected_sha256 = hashlib.sha256(encoded).hexdigest()
-    if destination.exists():
-        if destination.read_bytes() != encoded:
+    if os.path.lexists(destination):
+        if _read_existing_regular(destination) != encoded:
             raise FileExistsError(
                 "refusing to overwrite an existing non-identical GDW artifact"
             )
+        _fsync_directory(owner_root)
         return {
             "status": "EXPORTED",
             "path": str(destination),
@@ -101,6 +157,7 @@ def _export_json_artifact_unlocked(
             "reused": True,
             "immutable": True,
             "owner_scope": owner_scope,
+            "publication_mode": "REUSED",
         }
     owner_limit = _bounded_artifact_limit(
         "GDW_OWNER_MAX_ARTIFACTS", default=10_000, maximum=100_000
@@ -116,32 +173,57 @@ def _export_json_artifact_unlocked(
         raise RuntimeError("per-owner artifact quota exceeded")
     if sum(1 for _ in root.glob("*/*.json")) >= global_limit:
         raise RuntimeError("global artifact quota exceeded")
-    handle, temporary = tempfile.mkstemp(
-        prefix=".gdw-artifact-", suffix=".tmp", dir=owner_root
+    # A unique stage is owned only by this publisher. Process death can leave
+    # that hidden file unreferenced, but no retry or replica may delete an
+    # unknown concurrent stage or expose it as the final JSON artifact.
+    handle, staging = tempfile.mkstemp(
+        prefix=".gdw-artifact-",
+        suffix=".tmp",
+        dir=owner_root,
     )
+    reused = False
+    publication_mode = "HARD_LINK"
     try:
         with os.fdopen(handle, "wb") as stream:
             stream.write(encoded)
             stream.flush()
             os.fsync(stream.fileno())
-        try:
-            os.link(temporary, destination)
-        except FileExistsError:
-            if destination.read_bytes() != encoded:
-                raise FileExistsError(
-                    "refusing to overwrite a concurrently created "
-                    "non-identical GDW artifact"
-                )
+        for attempt in range(_LINK_MAX_ATTEMPTS):
+            try:
+                os.link(staging, destination)
+            except FileExistsError:
+                if _read_existing_regular(destination) != encoded:
+                    raise FileExistsError(
+                        "refusing to overwrite a concurrently created "
+                        "non-identical GDW artifact"
+                    )
+                reused = True
+                publication_mode = "REUSED"
+                break
+            except OSError as exc:
+                if (
+                    exc.errno not in _TRANSIENT_LINK_ERRNOS
+                    or attempt + 1 >= _LINK_MAX_ATTEMPTS
+                ):
+                    raise
+                publication_mode = "HARD_LINK_AFTER_FLUSH"
+                time.sleep(_LINK_RETRY_SECONDS)
+            else:
+                break
+        _fsync_directory(owner_root)
     finally:
-        if os.path.exists(temporary):
-            os.unlink(temporary)
+        try:
+            os.unlink(staging)
+        except FileNotFoundError:
+            pass
     return {
         "status": "EXPORTED",
         "path": str(destination),
         "sha256": expected_sha256,
-        "reused": False,
+        "reused": reused,
         "immutable": True,
         "owner_scope": owner_scope,
+        "publication_mode": publication_mode,
     }
 
 

gdw_runtime.py

@@ -442,8 +442,14 @@ def drain_once(
         "exported": exported,
         "failed": failed,
         "pending_effects": integrity["pending_effects"],
+        "claimed_effects": integrity["claimed_effects"],
+        "dead_letter_effects": integrity["dead_letter_effects"],
         "legacy_pending_proofs": integrity["pending_proofs"],
         "sqlite_integrity": integrity["sqlite_integrity"],
+        "invalid_effect_bindings": integrity["invalid_effect_bindings"],
+        "invalid_exported_artifacts": integrity[
+            "invalid_exported_artifacts"
+        ],
         "errors": errors,
     }
 
@@ -555,7 +561,28 @@ class OutboxSupervisor:
                         lease_seconds=self.lease_seconds,
                         worker_id=self.worker_id,
                     )
-                    if report["failed"] or report["legacy_pending_proofs"]:
+                    terminal_failure = (
+                        report["dead_letter_effects"]
+                        or report["sqlite_integrity"] != "ok"
+                        or report["invalid_effect_bindings"]
+                        or report["invalid_exported_artifacts"]
+                    )
+                    retryable_work = (
+                        report["failed"]
+                        or report["pending_effects"]
+                        or report["legacy_pending_proofs"]
+                    )
+                    if terminal_failure:
+                        delay = self.retry_max_seconds
+                        _set_drain_state(
+                            last_outcome="FAILED",
+                            last_error=(
+                                "bounded drain pass reported terminal "
+                                "integrity or dead-letter failures"
+                            ),
+                            last_report=report,
+                        )
+                    elif retryable_work:
                         retry_delay = min(
                             self.retry_max_seconds,
                             max(self.interval_seconds, retry_delay * 2),
@@ -564,8 +591,7 @@ class OutboxSupervisor:
                         _set_drain_state(
                             last_outcome="RETRY_SCHEDULED",
                             last_error=(
-                                "bounded drain pass reported failures or "
-                                "unmigrated legacy proofs"
+                                "bounded drain pass remains non-quiescent"
                             ),
                             last_report=report,
                         )

routers/gdw_frontier.py

@@ -286,6 +286,17 @@ def _write_readiness(
             blockers.append("OUTBOX_SUPERVISOR_NOT_RUNNING")
         if drain.get("last_outcome") != "SUCCEEDED":
             blockers.append("OUTBOX_SUPERVISOR_NOT_HEALTHY")
+        last_report = drain.get("last_report")
+        if isinstance(last_report, dict) and (
+            last_report.get("failed")
+            or last_report.get("pending_effects")
+            or last_report.get("dead_letter_effects")
+            or last_report.get("legacy_pending_proofs")
+            or last_report.get("invalid_effect_bindings")
+            or last_report.get("invalid_exported_artifacts")
+            or last_report.get("sqlite_integrity") != "ok"
+        ):
+            blockers.append("OUTBOX_SUPERVISOR_NOT_QUIESCENT")
         if drain.get("success_run_generation_id") != drain.get(
             "run_generation_id"
         ):
@@ -376,6 +387,34 @@ def _public_runtime_health(runtime: dict) -> dict:
             )
             if key in drain
         }
+        last_report = drain.get("last_report")
+        if isinstance(last_report, dict):
+            public_report = {
+                key: last_report.get(key)
+                for key in (
+                    "attempted",
+                    "exported",
+                    "failed",
+                    "pending_effects",
+                    "claimed_effects",
+                    "dead_letter_effects",
+                    "legacy_pending_proofs",
+                    "sqlite_integrity",
+                    "invalid_effect_bindings",
+                    "invalid_exported_artifacts",
+                )
+                if key in last_report
+            }
+            errors = last_report.get("errors")
+            if isinstance(errors, list):
+                public_report["errors"] = [
+                    value
+                    for value in errors
+                    if isinstance(value, str)
+                    and len(value) <= 96
+                    and re.fullmatch(r"[a-z_]+:[A-Za-z_]+", value)
+                ]
+            public_drain["last_report"] = public_report
     return {
         "startup_state": runtime.get("startup_state"),
         "evidence_label": runtime.get("evidence_label"),