revert(skip-unchanged): drop in-engine incremental Phase 1a

archive.py

@@ -109,20 +109,6 @@ def load_config(config_path):
             f"got: {config['cleanup_zip_on_success']!r}"
         )
 
-    # skip_unchanged: when true, a target whose live source is the same set as its
-    # last SUCCESSFUL archive — same file count, nothing modified since, same total
-    # bytes (exactly reclaim.verify's no-drift conditions) — is SKIPPED: no new
-    # tar, no new run log, so the prior success log stays the canonical record
-    # reclaim/coverage read. Default false: the unchanged/changed decision is still
-    # computed and logged (free measurement), but the archive runs anyway. The gate
-    # is bypassed when --fresh / --force-resume is given (an explicit re-archive).
-    config.setdefault("skip_unchanged", False)
-    if not isinstance(config["skip_unchanged"], bool):
-        raise ValueError(
-            f"config 'skip_unchanged' must be true or false, "
-            f"got: {config['skip_unchanged']!r}"
-        )
-
     return config
 
 
@@ -315,98 +301,6 @@ def enumerate_source(source_folder, file_pattern):
     return count, newest_mtime
 
 
-def skippable_prior_log(log_dir, project_name):
-    """The prior successful archive log to skip against — or None if skipping is
-    unsafe.
-
-    Returns the latest SUCCESSFUL run-log for project_name ONLY IF it is also the
-    newest run overall. Two deliberate properties:
-
-      * "Latest" is by parsed run_timestamp, NOT by log-file mtime — so a log
-        whose mtime was rewritten out of order (FS migration, cp without -p,
-        snapshot rehydrate) can't pick the wrong canonical archive, and the
-        choice agrees with coverage.classify (which also keys on run_timestamp).
-      * If the NEWEST run is not that success — i.e. a later run is `running` or
-        `failed` (a crash/retry is in flight) — return None so the gate does NOT
-        skip; the in-progress recovery must be allowed to re-establish a success.
-
-    Matches the log's exact 'project' field (not just the filename prefix) so a
-    sibling target sharing this prefix — e.g. '<proj>_volatile' vs '<proj>' —
-    never masquerades as this project's log.
-    """
-    import reclaim
-    newest, newest_dt = None, None
-    success, success_dt = None, None
-    for path in reclaim.find_logs(log_dir, project=project_name):
-        try:
-            log = reclaim.load_json(path)
-        except (ValueError, OSError):
-            continue
-        if log.get("project") != project_name:
-            continue
-        dt = reclaim.run_datetime(log.get("run_timestamp"))
-        if dt is None:
-            continue
-        if newest_dt is None or dt > newest_dt:
-            newest, newest_dt = log, dt
-        if log.get("status") == "success" and (success_dt is None or dt > success_dt):
-            success, success_dt = log, dt
-    # Skip only against a success that is the most recent run (no newer
-    # running/failed run shadowing it).
-    if success is not None and success is newest:
-        return success
-    return None
-
-
-def source_matches_log(log, source_folder, file_pattern):
-    """True if the live source is UNCHANGED versus this archive log.
-
-    Uses exactly the conditions reclaim.verify() uses to rule out DRIFT — same
-    file count, nothing modified after the archive ran, same total bytes — so the
-    skip decision can never disagree with reclaim's notion of "still archived".
-    One metadata-only walk (reclaim.scan_source); reads no file contents. Returns
-    False if the source is gone or the archive time can't be determined (treat as
-    changed → re-archive, the safe default).
-
-    KNOWN LIMITATION (by design): this stat-only triple cannot see an in-place
-    rewrite that preserves both byte length AND mtime — i.e. an equal-length
-    content swap with a preserved/backdated mtime. Gating a *write* on a
-    *detection* heuristic means such a change, if it ever occurred, would never be
-    re-archived while skip_unchanged is on. That is an accepted trade: detecting
-    it would require re-reading every byte each run (defeating the point), and
-    bit-level integrity of already-archived data is the separate job of
-    fs-checksum-sweep (#60), which re-hashes from bytes on a cadence. Research
-    data here is write-once, so the case is pathological; fs-checksum-sweep is the
-    backstop if it ever isn't.
-    """
-    import reclaim
-    # Only a log for the SAME target definition is comparable. If the archive's
-    # scope changed — a different file_pattern (selects a different set) or a
-    # re-pointed source_folder — the prior count/bytes describe a different
-    # selection, so re-archive under the new definition rather than skip.
-    if (log.get("file_pattern") or "") != (file_pattern or ""):
-        return False
-    log_src = log.get("source_folder")
-    if log_src and os.path.normpath(log_src) != os.path.normpath(source_folder):
-        return False
-    cutoff = reclaim.run_datetime(log.get("run_timestamp"))
-    if cutoff is None:
-        return False
-    scan = reclaim.scan_source(source_folder, file_pattern, cutoff)
-    if scan is None:
-        return False
-    live_count, newer_count, _samples, live_bytes = scan
-    archived_count = len(log.get("source_files") or {})
-    archived_bytes = log.get("source_bytes")
-    if live_count != archived_count:
-        return False
-    if newer_count > 0:
-        return False
-    if archived_bytes is not None and live_bytes != archived_bytes:
-        return False
-    return True
-
-
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
                    compression="deflate", allow_empty_files=False):
     """
@@ -968,34 +862,6 @@ def send_email(subject, body):
     allow_empty_files = config["allow_empty_files"]
     cleanup_zip_on_success = config["cleanup_zip_on_success"]
 
-    # ── Skip-unchanged gate ───────────────────────────────────────────────────
-    # If this target's live source is the same set as its last SUCCESSFUL archive
-    # (reclaim's no-drift conditions: same count, nothing newer, same bytes),
-    # there is nothing new to send to tape — re-archiving would only write a
-    # duplicate tar + a new run log. The decision is ALWAYS computed and logged
-    # (free measurement of redundancy); it is ACTED ON (skip, writing NO tar and
-    # NO new run log so the prior success log stays canonical for reclaim/coverage)
-    # only when skip_unchanged is enabled and the user did not force a re-archive
-    # with --fresh / --force-resume. Runs before any preflight so an unchanged
-    # target costs only one metadata walk.
-    _forced = (not resume_enabled) or force_resume
-    _prior = skippable_prior_log(log_dir, project_name)
-    if _prior is not None and source_matches_log(_prior, source_folder, file_pattern):
-        _since = _prior.get("run_timestamp")
-        if config["skip_unchanged"] and not _forced:
-            print(f"\n[skip-unchanged] {project_name}: unchanged since {_since} — "
-                  f"skipping (no tar, no new log; prior archive remains canonical).")
-            sys.exit(0)
-        elif config["skip_unchanged"] and _forced:
-            print(f"\n[skip-unchanged] {project_name}: unchanged since {_since}, but "
-                  f"--fresh/--force-resume forces a re-archive.")
-        else:
-            print(f"\n[skip-unchanged] {project_name}: WOULD skip (unchanged since "
-                  f"{_since}); skip_unchanged disabled — archiving anyway.")
-    elif _prior is not None:
-        print(f"\n[skip-unchanged] {project_name}: changed since "
-              f"{_prior.get('run_timestamp')} — archiving.")
-
     # `run` abstracts how a worker function is executed: in-process by default, or
     # submitted to the endpoint and waited on with --globus. Step 1/Step 2 call run()
     # and are identical in both modes.