Add accept_existing mode for incremental model migration

When accept_existing=True, the build keeps cached artifacts even if
the transform fingerprint changed (e.g. model swap). Only artifacts
with new/changed inputs get rebuilt with the new config.

- Skip transform fingerprint comparison in _layer_fully_cached
- Skip per-artifact fingerprint check in _save_artifact
- Accept any existing artifact in cached_by_inputs lookup
- Wire through MeshConfig.server.accept_existing → run()

Author: marklubin

src/synix/build/runner.py

@@ -106,6 +106,7 @@ def run(
     progress=None,
     validate: bool = False,
     error_classifier: ErrorClassifier | None = None,
+    accept_existing: bool = False,
 ) -> RunResult:
     """Execute the full pipeline — walk DAG, run transforms, materialize projections.
 
@@ -120,6 +121,10 @@ def run(
             (default), all LLM errors are fatal and abort the build. Pass
             ``LLMErrorClassifier()`` to enable DLQ — content-filter and
             input-too-large errors will be skipped instead of aborting.
+        accept_existing: When True, keep existing cached artifacts even if the
+            transform fingerprint changed (e.g. model swap). Only rebuilds
+            artifacts with new/changed inputs. Useful for incremental model
+            migration — existing work is preserved, new inputs use the new config.
 
     Returns:
         RunResult with build statistics.
@@ -199,7 +204,7 @@ def run(
             layer_built: list[Artifact] = []
             dlq_before = len(result.dlq)
 
-            if _layer_fully_cached(layer, inputs, store, transform_fp):
+            if _layer_fully_cached(layer, inputs, store, transform_fp, accept_existing=accept_existing):
                 # All cached — load existing artifacts
                 existing = store.list_artifacts(layer.name)
                 for art in existing:
@@ -226,17 +231,24 @@ def _save_artifact(
                     _layer=layer,
                     _transform_fp=transform_fp,
                     _inputs=inputs,
+                    _accept_existing=accept_existing,
                 ) -> None:
                     effective_inputs = parent_inputs if parent_inputs is not None else _inputs
                     # Compute per-artifact build fingerprint
                     build_fp = compute_build_fingerprint(_transform_fp, artifact.input_ids)
 
-                    rebuild, _reasons = needs_rebuild(
-                        artifact.label,
-                        artifact.input_ids,
-                        store,
-                        current_build_fingerprint=build_fp,
-                    )
+                    # In accept_existing mode, skip fingerprint comparison —
+                    # only rebuild if the artifact doesn't exist at all.
+                    if _accept_existing:
+                        existing = store.load_artifact(artifact.label)
+                        rebuild = existing is None
+                    else:
+                        rebuild, _reasons = needs_rebuild(
+                            artifact.label,
+                            artifact.input_ids,
+                            store,
+                            current_build_fingerprint=build_fp,
+                        )
                     if rebuild:
                         artifact.metadata["layer_name"] = _layer.name
                         artifact.metadata["layer_level"] = _layer._level
@@ -288,12 +300,17 @@ def _on_batch_complete(artifacts: list[Artifact], unit_inputs: list[Artifact]) -
                 existing_artifacts = store.list_artifacts(layer.name)
                 cached_by_inputs: dict[tuple[str, ...], list[Artifact]] = {}
                 for art in existing_artifacts:
-                    stored_tfp_data = art.metadata.get("transform_fingerprint")
-                    if stored_tfp_data is not None:
-                        stored_fp = Fingerprint.from_dict(stored_tfp_data)
-                        if transform_fp.matches(stored_fp):
-                            key = tuple(sorted(art.input_ids))
-                            cached_by_inputs.setdefault(key, []).append(art)
+                    if accept_existing:
+                        # Accept any existing artifact regardless of fingerprint
+                        key = tuple(sorted(art.input_ids))
+                        cached_by_inputs.setdefault(key, []).append(art)
+                    else:
+                        stored_tfp_data = art.metadata.get("transform_fingerprint")
+                        if stored_tfp_data is not None:
+                            stored_fp = Fingerprint.from_dict(stored_tfp_data)
+                            if transform_fp.matches(stored_fp):
+                                key = tuple(sorted(art.input_ids))
+                                cached_by_inputs.setdefault(key, []).append(art)
 
                 def _on_cached(cached_arts: list[Artifact], unit_inputs: list[Artifact]) -> None:
                     for cached_art in cached_arts:
@@ -658,14 +675,15 @@ def _layer_fully_cached(
     inputs: list[Artifact],
     store: SnapshotArtifactCache,
     transform_fp: Fingerprint | None = None,
+    accept_existing: bool = False,
 ) -> bool:
     """Check if a layer can be entirely skipped (all artifacts cached)."""
     existing = store.list_artifacts(layer.name)
     if not existing:
         return False
 
-    # Check transform identity via fingerprint
-    if transform_fp is not None:
+    # Check transform identity via fingerprint (skip when accept_existing)
+    if transform_fp is not None and not accept_existing:
         for art in existing:
             stored_tfp_data = art.metadata.get("transform_fingerprint")
             if stored_tfp_data is None:

src/synix/mesh/config.py

@@ -36,6 +36,7 @@ class ServerConfig:
     build_min_interval: int = 300
     build_quiet_period: int = 60
     build_max_delay: int = 1800
+    accept_existing: bool = False
 
 
 @dataclass
@@ -185,6 +186,7 @@ def load_mesh_config(path: Path) -> MeshConfig:
         build_min_interval=srv.get("build_min_interval", ServerConfig.build_min_interval),
         build_quiet_period=quiet_period,
         build_max_delay=srv.get("build_max_delay", ServerConfig.build_max_delay),
+        accept_existing=srv.get("accept_existing", ServerConfig.accept_existing),
     )
 
     # Client config

src/synix/mesh/server.py

@@ -509,7 +509,10 @@ async def _run_build() -> None:
                 # 3. Run build in a thread (sync build system)
                 build_phase = "building"
                 pipeline.build_dir = str(build_dir)
-                result = await asyncio.to_thread(run, pipeline, source_dir=source_dir)
+                result = await asyncio.to_thread(
+                    run, pipeline, source_dir=source_dir,
+                    accept_existing=config.server.accept_existing,
+                )
 
                 async with _state_lock:
                     build_count += 1