Make Clustrix a data mover: stage declared inputs and outputs (reverses the "not a data mover" limitation)

## Summary

`docs/source/introduction.rst:115` currently declares:

> **It is not a data mover.** Clustrix ships your *code and arguments*, not your dataset. If your function needs a 200 GB file, that file has to already be reachable from the worker. The filesystem utilities help you inspect and locate remote data, but they are not a transfer service for bulk inputs.

That is an accurate description of current *behaviour* and an inaccurate description of *intent*. The record shows data movement was an architectural goal, was designed, was partially built, and was abandoned mid-wiring — after which the docs described the abandonment as a deliberate scope boundary.

Full design: `notes/data-mover-design.md`. This issue carries enough to act on without it.

---

## 1. The conflicting design element (found, quoted)

**Issue #64** (CLOSED), "Core Architecture: Function Serialization and Dependency Management for Remote Execution", lists under *Current Approach Limitations* — i.e. as a defect to fix:

> 3. **No local file support** - cannot access local modules, data files, or custom code

and under *Proposed Solution Direction*:

> ### Dependency Packaging Approach
> 1. **Dependency Detection**: Analyze function for local imports, file references, and dependencies
> 2. **File Packaging**: Create zip archive of all local dependencies
> 3. **Remote Deployment**: SCP and unpack files on cluster with unique identifier (MD5 hash)
> 4. **Environment Recreation**: Patch import paths and execute function in recreated context

It even names the hard parts: *"How to maintain security while allowing file transfer?"*, *"Network transfer overhead for large dependencies"*, *"File system security considerations"*.

#64 was closed on the strength of `docs/function_serialization_technical_design.md`, whose final revision (commit `3417b84`, "Update technical design document - IMPLEMENTATION COMPLETE") claims:

> ### ✅ Phase 2: Dependency Packaging Core (COMPLETE)
> - ✅ Create `FilePackager` for selective file collection
> - ✅ Build package deployment system for remote transfer
>
> ### ✅ Phase 4: Integration & Testing (COMPLETE)
> - ✅ Integrate with existing `ClusterExecutor`

**That last bullet is false** — see §2. The doc was deleted in `fb373d2`.

**Supporting element, issue #20** ("Proposed classes"), the original `Job` spec:

> - stores a function, arguments, and pointers to the appropriate file paths (e.g., data, results, scratch)

### The part of the record that cuts the other way — report honestly

Issue #10, comment by @jeremymanning:

> Syncing *code* seems like a great idea-- we want to be able to ensure that the user is running what they think they're running, and managing file transfers would be fantastic. Syncing *results* or *data* seems like a bad idea; I'm imagining that could eat up some serious space on a laptop hard drive that might not have enough room.

So the original position was **asymmetric**: upload was wanted ("managing file transfers would be fantastic"); *download* of data/results was explicitly rejected on grounds of local disk and transfer time. #64 later widened "code" to "local modules, data files, or custom code" without revisiting #10.

**Synthesis this plan adopts:** the upload direction was always in scope and was never delivered; the download direction was deliberately out of scope and that objection is still valid. The feature is therefore **not symmetric**. Outputs come back by explicit request only, never by inference.

---

## 2. What already exists (file:line)

**Transport exists and is wired** — but only for the payload:
- `clustrix/executor_connections.py:146` `upload_file` (`sftp.put`), `:156` `download_file` (`sftp.get`). No chunking, progress, or resume.
- Callers: `executor_schedulers.py:99` (`function_data.pkl` up), `executor_core.py:197` (`result.pkl` down), `executor_scheduler_status.py:473` (error pickle down).

**Detection exists and is completely orphaned:**
- `clustrix/dependency_analysis.py:249` `_analyze_file_references` finds paths three ways: string-literal first arg to a call in `self.file_operations = {"open","read","write","load","dump","save"}` (line 140); any `read/write/readline/writelines` method call, recorded as `path=""`; and **any string constant containing a separator and ending in one of 17 hardcoded extensions**. `add_file_references` (line 110) promotes every relative hit into `data_files` (line 99).
- `clustrix/file_packaging.py:319` `_add_data_files` zips them into `data/`.
- **Nothing calls it.** `grep -rn "FilePackager\|package_function" clustrix/decorator.py clustrix/executor_core.py clustrix/executor_connections.py clustrix/utils.py` → empty. The only importers of `file_packaging.py` are `clustrix/__init__.py:39` and `tests/test_file_packaging.py`. Same class of finding as #122.

**The `cluster_*` API is read-only:** `clustrix/filesystem.py:567-660` defines exactly `cluster_ls/find/stat/exists/isdir/isfile/glob/du/count_files`. No `cluster_put`, no `cluster_get`. `ClusterFilesystem` holds an SFTP client (`filesystem.py:225`) used only for `stat`-style reads.

**Shared-filesystem detection already exists:** `filesystem.py:113` `_auto_detect_cluster_location`, requiring `same_host and shared_filesystem`. Its comment documents a real past bug (a laptop on the VPN judged to *be* the cluster) — worth preserving as a warning.

**Cleanup:** `config.py:86` `cleanup_on_success=True`; `executor_core.py:214` runs `rm -rf {remote_dir}`. It knows only about the job directory.

**Inbound-byte security:** `executor_core.py:131` `_verify_result_signature` reads `result.pkl.hmac` (line 160); `executor_core.py:204` calls it before `dill.loads` at line 211.

**Prior art for out-of-band staging:** `clustrix/hf_jobs.py:352-375` already stages oversized payloads to a private HF dataset repo, docstring: *"HuggingFace rejects very large environment variables, which capped a job's arguments at a few hundred kilobytes -- fine for a function, useless for data."*

### Gap table

| Capability | State |
|-|-|
| SFTP put/get | Exists, wired |
| Detect referenced files | Exists, orphaned, heuristic |
| Package data files | Exists, orphaned |
| Shared-FS detection | Exists, `cluster_*` only |
| Explicit put/get API | **Missing** |
| Content-addressed dedup / manifest | **Missing** |
| Size policy (refuse/warn/delegate) | **Missing** |
| Output return path | **Missing** |
| Resume / integrity on partial transfer | **Missing** |
| Staged-input lifecycle vs `cleanup_on_success` | **Missing** |

---

## 3. Design decisions

**Semantics — explicit declaration, never inference.** The existing detector's third rule is "any string ending in one of 17 extensions", so automatic staging would upload on a literal like `"s3://bucket/notes.log"`. Silent inferred I/O is the worst failure mode here. API:

```python
@cluster(cores=8,
inputs=["data/subjects.h5", "config/model.yaml"],
outputs=["results/*.npz"])
def fit(subject): ...
```

Inputs are visible to the function at **the same relative path they had locally**, so the function is unchanged between local and remote runs. Outputs are fetched only on success, only when declared. Primitives underneath: `cluster_put(local, remote, config)` / `cluster_get(remote, local, config)` in `filesystem.py`.

**Size — three bands, refuse past the top.** `stage_warn_bytes` (default 100 MB) → silent; below `stage_max_bytes` (default 5 GB) → transfer with a warning naming file, size, throughput, ETA; at/above → **raise**, naming the file, the threshold, the config key, and the two remedies (shared storage; `stage_backend="rsync"`). A refusal beats a silent multi-hour SFTP that looks like a hang. `stage_backend ∈ {"sftp","rsync"}`, opt-in only — auto-selecting by size makes the transport depend on the data.

**Idempotence — content hash with an mtime+size fast path.** `(size, mtime_ns, inode)` hit in the local cache reuses the cached BLAKE2b-256 digest; otherwise stream-hash. Blobs live content-addressed at `{remote_work_dir}/_stage//`, **outside** any job dir; job dirs get symlinks (→ hardlink → copy fallback). Remote manifest `_stage/manifest.json` (flock + atomic rename; `digest -> {size, first_seen, last_used, refcount}`) is authoritative. Local `~/.clustrix/stage-cache/.json` is a pure optimisation — deleting it costs time, never correctness.

**Shared filesystems — per-path probe, not per-host.** New `shared_filesystem_roots` config; if `realpath(local)` is under one **and** a one-shot remote `stat` matches size+mtime, stage nothing and point the job at the absolute path. The remote `stat` is mandatory — `filesystem.py:113`'s comment records what happens when this is answered by names.

**Cleanup — three distinct lifetimes.** Job dir: `cleanup_on_success`, unchanged. Staged blobs: survive the job, reclaimed by a submission-time reaper via `stage_cache_ttl_days` (7) / `stage_cache_max_bytes`, plus explicit `clustrix.clear_stage_cache(config)` and a CLI verb. Fetched outputs: **never** touched by clustrix.

**Security — downloaded data files do NOT need an HMAC, and here is why.** `result.pkl` is signed because *clustrix itself* calls `dill.loads` on it; the RCE path is created by our deserialization, not by the transfer. A staged output is written to disk and handed over as a path — clustrix never interprets it, and the attacker who could tamper with it already controls the remote account. What we owe instead: (1) **integrity** — digest every staged input, re-verify on the worker; digest outputs on the worker, verify after fetch; (2) **path confinement** — output globs `realpath`-confined to the job dir, escapes rejected not clamped (Zip-Slip class); (3) `0600`/`0700` modes, matching the care in `executor_connections.py::create_remote_file`; (4) **refuse credential-shaped paths** (`~/.ssh/*`, `*.pem`, `.env`) without an explicit override; (5) the manifest is remote-origin data — fixed-schema `json.load`, never pickle/eval.

**Failure modes:** upload to `.partial` + fsync + size check + atomic rename, never register an unverified digest; quota exceeded → reap, retry once, then raise naming the store and `clear_stage_cache` (never silently run the job without the file); permission denied → probe writability once per session and fail before any bytes move; file changed mid-upload → re-`stat` after transfer and raise if `(size, mtime_ns)` moved; missing declared input → hard failure at submission; missing declared output → warn (or error under `require_outputs=True`); concurrent stagers → per-digest lock, loser waits for the rename; no symlink support on the worker → hardlink → copy, logged (a copy doubles quota).

**Out of scope** (this is what keeps it from becoming a workflow engine): no sync/mirror/watch, no DAG or data-derived ordering, no cluster→cluster transfer, no S3/GCS backends in the transfer path, no compression/format conversion, no provenance DB, and **no automatic inference of inputs from source in any phase**.

---

## 4. Phases

Verification targets are the backends this project has actually proven: real SLURM, real SSH GPU host, HF Jobs. Evidence lands in `docs/evidence/`.

### Phase 1 — Primitives: `cluster_put` / `cluster_get`
Add both to `filesystem.py` alongside the existing nine, over the existing SFTP client (`filesystem.py:225`) / `shutil` locally. Streamed BLAKE2b, atomic `.partial`+rename, `0600`/`0700`, path confinement, size bands with `stage_warn_bytes`/`stage_max_bytes` on `ClusterConfig`. No manifest, no caching, no decorator changes.

**DoD:** a file `cluster_put` to the real SLURM host *and* the real SSH GPU host, digest verified there by remote command; `cluster_get` round-trips byte-identical; a file over `stage_max_bytes` raises with the documented message; a killed transfer leaves only a `.partial` with nothing registered; unit tests for confinement and mode; evidence committed.

### Phase 2 — Content-addressed store, manifest, dedup
`_stage/` layout, locked+atomic remote manifest, local `(path,size,mtime,inode)→digest` cache, skip-if-present, symlink/hardlink/copy materialisation, per-digest lock.

**DoD:** on real SLURM, staging the same ~1 GB file twice transfers bytes exactly **once** (measured); deleting the local cache still yields zero re-upload; two concurrent processes produce one copy and one manifest entry; a corrupted manifest is detected and rebuilt from store contents rather than crashing.

### Phase 3 — `@cluster(inputs=..., outputs=...)`
Resolve declarations at submission, stage, materialise at original relative paths, run, collect declared outputs on success, fetch with digest verification. `require_outputs`.

**DoD:** an unmodified function opening `"data/x.csv"` relatively runs identically locally and on real SLURM with `inputs=["data/x.csv"]`; glob outputs return and verify; a failed job fetches nothing; an output pattern escaping the job dir is rejected on the worker; `cleanup_on_success` removes the job dir while `_stage/` survives — proven by a second run reusing the blob.

### Phase 4 — Shared-FS elision and quota safety
`shared_filesystem_roots`, per-path remote `stat` probe, reaper (`stage_cache_ttl_days`, `stage_cache_max_bytes`), `clear_stage_cache`, quota retry-once, CLI verb.

**DoD:** on real SLURM a file already on the site's shared root stages with **zero** bytes transferred and the job reads the original path; the negative case (same filename, different content, not actually shared) is correctly *not* elided; a store over `stage_cache_max_bytes` is reaped below it on next submission with the eviction logged; an induced quota failure produces the documented error, not a hang.

### Phase 5 — `stage_backend="rsync"` (only if 1-4 hit a wall)
`rsync -az --partial --info=progress2` over the existing SSH identity, same manifest, same digests, strictly opt-in.

**DoD:** a multi-GB stage to real SLURM completes via rsync with the same digest the SFTP path would produce, resumes after an induced interruption, and a host without `rsync` errors clearly rather than falling back.

---

## 5. Documentation

`docs/source/introduction.rst:115` is owned by a separate docs sweep — do not edit it here.

**Interim wording:**
> **It does not yet move your data.** Clustrix ships your *code and arguments*, not your dataset. If your function needs a 200 GB file, that file has to already be reachable from the worker today. The filesystem utilities help you inspect and locate remote data, but they are not yet a transfer service for bulk inputs. Staging declared inputs and outputs is planned — see this issue.

**After Phases 1-4**, it leaves the "what it is not" list and becomes:
> **It moves the data you declare.** `@cluster(inputs=[...], outputs=[...])` stages the files your function names, content-addressed and deduplicated, so re-running does not re-upload unchanged inputs and a file already on shared storage is not uploaded at all. It is not a sync tool: nothing moves that you did not name, and nothing comes back that you did not ask for.

The "not a workflow engine" and "not a low-latency dispatcher" bullets stay; the out-of-scope list above is what keeps them true.

---

## 6. Decision needed before any code is written

**Is the stable path contract "the same relative path the file had locally", or "an opaque staged path handed to the function"?**

Relative-path is what lets a function run unmodified in both places — the entire value proposition — but it asserts control over the worker's working-directory layout, breaks for absolute-path inputs, and forces the symlink/hardlink/copy machinery in Phase 2. Opaque-path is far simpler and impossible to get wrong, but every user rewrites their function to take a path argument, at which point they may reasonably ask what this bought them over `scp`.

This plan assumes relative-path. If that is wrong, **Phases 2 and 3 both shrink substantially** and must be re-scoped first.

## Related
- #64 (closed; the conflicting design element)
- #20 (`Job` holds "pointers to the appropriate file paths (e.g., data, results, scratch)")
- #10 (the contrary position on syncing data/results)
- #122 (orphaned modules — `file_packaging.py` is one of them; this issue would give it a purpose or replace it)

Start with clustrix/filesystem.py, especially the existing ClusterFilesystem SFTP client around line 225 and the cluster_* methods at lines 567-660. Implement and verify Phase 1's primitives first, then run the relevant unit tests for confinement, modes, size limits, and interrupted transfers. Done includes real SLURM and SSH GPU-host round trips, byte verification, and evidence in docs/evidence/.

Written by the indexing model from the issue text.