feat(sync): add verifiable global-sync foundation

.github/workflows/unit-tests.yml

@@ -29,7 +29,8 @@ jobs:
     if: github.event.pull_request.draft != true
     name: Run Unit Tests
     runs-on: ubuntu-latest
-    timeout-minutes: 30
+    # The measured green broad suite takes ~30 minutes; retain bounded 50% headroom.
+    timeout-minutes: 45
 
     env:
       PDD_PATH: ${{ github.workspace }}/pdd
@@ -55,6 +56,83 @@ jobs:
       - name: Install dependencies
         run: pip install -e ".[dev]" pytest-timeout
 
+      - name: Provision and verify protected Linux sandbox
+        run: |
+          sudo apt-get update
+          sudo apt-get install --yes bubblewrap
+          private_root="$(mktemp -d)"
+          chmod 700 "$private_root"
+          mkdir -m 700 "$private_root/scratch"
+          PRIVATE_ROOT="$private_root" python - <<'PY'
+          import os
+          import json
+          import subprocess
+          import sys
+          from pathlib import Path
+          from pdd.sync_core.supervisor import (
+              _sandbox_command,
+              released_runtime_closure_paths,
+              run_supervised,
+          )
+
+          root = Path(os.environ["PRIVATE_ROOT"])
+          resolved_interpreter = Path(sys.executable).resolve()
+          print("resolved interpreter:", resolved_interpreter)
+          print(subprocess.run(
+              ["ldd", str(resolved_interpreter)], check=False, text=True,
+              capture_output=True,
+          ).stdout)
+          sandbox, _ = _sandbox_command(
+              [sys.executable, "-c", "import math"], (root / "scratch",)
+          )
+          print("sandbox destinations:", json.loads(sandbox[-2]))
+          print("runtime native paths:", [
+              str(path) for label, path in released_runtime_closure_paths()
+              if label.startswith("native/")
+          ])
+          script = """from pathlib import Path
+          import math
+          Path('ok').write_text('ok')
+          try:
+              Path('../outside').write_text('forbidden')
+          except OSError:
+              pass
+          else:
+              raise SystemExit('write escaped private scratch root')
+          """
+          result, surviving = run_supervised(
+              [sys.executable, "-c", script],
+              cwd=root / "scratch", timeout=10, env=dict(os.environ),
+              writable_roots=(root / "scratch",),
+          )
+          assert result.returncode == 0, result.stderr
+          assert not surviving
+          assert (root / "scratch" / "ok").read_text() == "ok"
+          assert not (root / "outside").exists()
+          PY
+
+      - name: Verify protected pytest smoke
+        run: >
+          pytest -q
+          tests/test_sync_core_runner.py::test_passing_collected_test_is_pass
+          --timeout=60
+
+      - name: Verify protected signer containment smoke
+        run: >
+          pytest -q
+          tests/test_sync_core_trust.py::test_remote_signer_timeout_reaps_env_cleared_detached_descendant
+          --timeout=60
+
+      - name: Run focused protected-runner tests
+        run: >
+          pytest -q
+          tests/test_sync_core_supervisor.py
+          tests/test_sync_core_runner.py
+          tests/test_sync_core_trust.py
+          tests/test_sync_core_lifecycle_scenarios.py
+          tests/test_sync_core_reporting.py
+          --timeout=60
+
       - name: Validate architecture vs prompt includes (fixture smoke)
         run: pdd checkup --validate-arch-includes --project-root tests/fixtures/arch_include_validate_ok
 

.gitignore

@@ -242,6 +242,7 @@ yarn.lock
 # PDD internal directory (backups, locks, worktrees, etc.)
 .pdd/*
 !.pdd/
+!.pdd/repository-id
 !.pdd/meta/
 .pdd/meta/*.json
 !.pdd/meta/agentic_bug_orchestrator_python.json

.pdd/repository-id

@@ -0,0 +1 @@
+3b4d7b1c-d6cc-4752-ba93-6b98d1a710e0

pdd/auto_deps_main.py

@@ -48,6 +48,8 @@ def auto_deps_main(
         ("", 0.0, f"Error: {exc}") on non-Abort failures so orchestrators can
         continue gracefully.
     """
+    from .sync_core.finalize import preflight_legacy_mutation
+    preflight_legacy_mutation({"prompt": Path(prompt_file)})
     console = Console()
     quiet = ctx.obj.get("quiet", False) if ctx.obj else False
 
@@ -254,12 +256,18 @@ def auto_deps_main(
                             model=model_name,
                         )
                     except Exception as fp_exc:
+                        from .sync_core.finalize import CanonicalFinalizationError
+                        if isinstance(fp_exc, CanonicalFinalizationError):
+                            raise
                         if not quiet:
                             console.print(
                                 f"[yellow]Warning: Failed to save fingerprint for "
                                 f"{basename}_{language}: {fp_exc}[/yellow]"
                             )
             except Exception as meta_exc:
+                from .sync_core.finalize import CanonicalFinalizationError
+                if isinstance(meta_exc, CanonicalFinalizationError):
+                    raise
                 # Never mask a successful auto-deps result on metadata errors
                 if not quiet:
                     console.print(
@@ -272,6 +280,9 @@ def auto_deps_main(
         # Re-raise to allow orchestrators (e.g. pdd sync) to stop the loop
         raise
     except Exception as exc:
+        from .sync_core.finalize import CanonicalFinalizationError
+        if isinstance(exc, CanonicalFinalizationError):
+            raise
         if not quiet:
             console.print(f"[red]Error in auto-deps: {exc}[/red]")
         return "", 0.0, f"Error: {exc}"

pdd/ci_drift_heal.py

@@ -11,6 +11,7 @@
 
 import argparse
 import csv
+import json
 import math
 import os
 import re
@@ -57,6 +58,46 @@ def _heal_subprocess_timeout() -> int:
 _INVARIANT_KEYS = {"include", "pdd_tags", "percent_markers", "fenced_blocks"}
 
 
+def _dry_run_json_summary(report: Dict[str, Any]) -> Dict[str, Any]:
+    """Return the fixed dry-run schema without paths or diagnostic payloads."""
+    summary = report.get("summary")
+    if not isinstance(summary, dict):
+        summary = {}
+
+    def count(name: str) -> int:
+        value = summary.get(name)
+        valid_count = (
+            isinstance(value, int) and not isinstance(value, bool) and value >= 0
+        )
+        return value if valid_count else 0
+
+    units = report.get("units")
+    if not isinstance(units, list):
+        units = []
+    projected_units = []
+    for unit in units:
+        if not isinstance(unit, dict):
+            continue
+        identity = {
+            name: unit.get(name)
+            for name in ("basename", "language", "classification")
+        }
+        if all(isinstance(value, str) and value for value in identity.values()):
+            projected_units.append(identity)
+
+    return {
+        "ok": report.get("ok") is True,
+        "consumer": "ci-heal",
+        "summary": {
+            "metadata_stale": count("metadata_stale"),
+            "conflicts": count("conflicts"),
+            "unbaselined": count("unbaselined"),
+            "failures": count("failures"),
+        },
+        "units": projected_units,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Data model
 # ---------------------------------------------------------------------------
@@ -299,25 +340,29 @@ def _discover_modules() -> List[Tuple[str, str, Any]]:
     """Discover (basename, language, prompt_path) tuples."""
     try:
         from pdd.user_story_tests import discover_prompt_files
-    except ImportError:
-        return []
+    except ImportError as exc:
+        raise RuntimeError("cannot import prompt discovery") from exc
     try:
         prompt_files = discover_prompt_files()
-    except Exception:
-        return []
+    except Exception as exc:
+        raise RuntimeError("prompt discovery failed") from exc
 
     try:
         from pdd.operation_log import infer_module_identity
-    except ImportError:
-        return []
+    except ImportError as exc:
+        raise RuntimeError("cannot import canonical module identity") from exc
 
     out: List[Tuple[str, str, Any]] = []
+    failures: List[str] = []
     for entry in prompt_files:
         try:
             basename, language = infer_module_identity(str(entry))
-        except Exception:
+        except Exception as exc:
+            failures.append(f"{entry}: {exc}")
             continue
         out.append((basename, language, entry))
+    if failures:
+        raise RuntimeError("module identity failed: " + "; ".join(failures))
     return out
 
 
@@ -412,6 +457,8 @@ def detect_drift(
         changed_files = _get_git_changed_files(diff_base)
 
     discovered = _discover_modules()
+    if not discovered and parsed is None:
+        raise RuntimeError("no synchronization units were discovered")
     if parsed is not None:
         wanted = set(parsed)
         discovered = [t for t in discovered if t[0] in wanted]
@@ -459,8 +506,10 @@ def detect_drift(
             decision = sync_determine_operation(
                 basename, language, target_coverage=90.0, log_mode=True
             )
-        except Exception:
-            continue
+        except Exception as exc:
+            raise RuntimeError(
+                f"classification failed for {basename}/{language}: {exc}"
+            ) from exc
 
         op = _extract_op(decision)
         if not op or op in ("nothing", "synced"):
@@ -1616,6 +1665,12 @@ def heal_module(drift: DriftInfo, env: Dict[str, str]) -> Optional[bool]:
             label=f"pdd sync {drift.basename}",
         )
         return ok if ok else False
+    if op == "fail_and_request_manual_merge":
+        console.print(
+            f"[yellow]manual merge required for {drift.basename}; "
+            "skipping auto-heal[/yellow]"
+        )
+        return None
 
     console.print(f"[red]unknown operation '{op}' for {drift.basename}[/red]")
     return False
@@ -1848,13 +1903,43 @@ def main(
     as_json: bool = False,
 ) -> int:
     """Detect drift, heal modules, and commit healed changes."""
+    from pdd.continuous_sync import canonical_sync_enabled
+
+    if canonical_sync_enabled(Path.cwd()) and not dry_run:
+        from pdd.sync_core import CanonicalReportOptions, build_canonical_report
+
+        base_ref = (
+            diff_base.split("...", 1)[0]
+            if diff_base and "..." in diff_base
+            else "HEAD"
+        )
+        try:
+            canonical = build_canonical_report(
+                _repo_root(),
+                CanonicalReportOptions(
+                    base_ref=base_ref,
+                    head_ref="HEAD",
+                    modules=tuple(modules or ()),
+                ),
+            )
+        except (OSError, RuntimeError, ValueError) as exc:
+            console.print(f"[red]canonical sync check failed: {exc}[/red]")
+            return 1
+        if not canonical["ok"]:
+            if as_json:
+                print(json.dumps(canonical, indent=2, sort_keys=True))
+            else:
+                console.print("[red]canonical sync predicate failed[/red]")
+                for error in canonical.get("errors", []):
+                    console.print(f"[red]- {error}[/red]")
+            return 1
+        return 0
     if dry_run:
         from pdd.continuous_sync import build_report
-        import json as _json
 
         report = build_report(consumer="ci-heal", modules=modules)
         if as_json:
-            print(_json.dumps(report, indent=2, sort_keys=True))
+            print(json.dumps(_dry_run_json_summary(report), indent=2, sort_keys=True))
         else:
             summary = report["summary"]
             console.print(

pdd/commands/__init__.py

@@ -24,6 +24,7 @@
 from .firecrawl import firecrawl_cache
 from .story import story_cli
 from .reconcile import install_hooks, reconcile
+from .sync_core import baseline, certify, recover, validate
 
 def register_commands(cli: click.Group) -> None:
     """Register all subcommands with the main CLI group."""
@@ -55,6 +56,10 @@ def register_commands(cli: click.Group) -> None:
     cli.add_command(which)
     cli.add_command(reconcile)
     cli.add_command(install_hooks)
+    cli.add_command(certify)
+    cli.add_command(recover)
+    cli.add_command(baseline)
+    cli.add_command(validate)
 
     # Register templates group directly to commands dict to handle nesting if needed,
     # or just add_command works for groups too.

pdd/commands/maintenance.py

@@ -363,6 +363,15 @@ def _restore_model_default(_prev=_prev_model_default):
     effective_compressed_context = _resolve_compressed_context(compressed_context)
     ctx.obj["compressed_context"] = effective_compressed_context
 
+    if not dry_run:
+        from ..sync_core.finalize import preflight_legacy_mutation
+
+        try:
+            preflight_legacy_mutation()
+        except RuntimeError as exc:
+            ctx.obj["_suppress_core_dump"] = True
+            raise click.UsageError(str(exc)) from exc
+
     # No basename -> global Tier 1 sync
     if basename is None:
         if snapshot_context:

pdd/commands/reconcile.py

@@ -75,7 +75,7 @@ def _emit_report(report: dict, as_json: bool) -> None:
     "--heal",
     is_flag=True,
     default=False,
-    help="Refresh valid stale fingerprints without LLM calls.",
+    help="Deprecated; blind fingerprint acceptance is disabled.",
 )
 @click.option(
     "--ledger",
@@ -103,10 +103,15 @@ def reconcile(
     ctx.ensure_object(dict)
     if as_json:
         ctx.obj["_suppress_result_summary"] = True
+    if heal:
+        raise click.UsageError(
+            "--heal is disabled because changing a fingerprint does not prove "
+            "semantic synchronization; run pdd sync/update/resolve instead"
+        )
     report = build_report(
         consumer="reconcile",
         modules=[module_name] if module_name else None,
-        heal=heal,
+        heal=False,
         ledger=ledger,
     )
     _emit_report(report, as_json)