From bd070b62e4c508406f4c67a3c2d2ebbcac0038ac Mon Sep 17 00:00:00 2001 From: POM Date: Sat, 30 May 2026 01:39:30 +0200 Subject: [PATCH 1/5] Merge remote dataset on upload instead of overwriting Replace the destructive wholesale-overwrite upload with a safe download -> union-merge -> re-redact -> reupload in push_to_huggingface: - merge_jsonl_union in jsonl_tools.py unions remote+local by (source, session_id) (avoids start_time/project key drift), keeps the superset on conflict, and guarantees merged_total >= remote_total so a remote shrink is impossible. - Carried-forward remote records are re-run through the CURRENT redaction pipeline (transform_session + Anonymizer, built exactly as the export loop does) so old-policy redaction is never re-published. Idempotent. - Download fails closed: only EntryNotFoundError/RepositoryNotFoundError count as "empty remote"; any other error aborts the push rather than overwriting the remote with a local-only file. - Reads remote records raw (preserves originalFile), uses parent_commit for optimistic concurrency with 412 retry, updates meta["sessions"] to the merged total. Tests: tests/test_jsonl_merge.py covers carry-forward, dedup-by-sid, superset selection, originalFile preservation, and re-redaction. Co-Authored-By: Claude Opus 4.8 (1M context) --- dataclaw/_cli/commands.py | 20 +- dataclaw/_cli/exporting.py | 141 ++++++++++- dataclaw/jsonl_tools.py | 165 +++++++++++++ ...liability-merge-and-privacy-filter-plan.md | 161 +++++++++++++ tests/test_cli_exporting.py | 115 ++++++++- tests/test_cli_facade.py | 3 +- tests/test_jsonl_merge.py | 218 ++++++++++++++++++ 7 files changed, 799 insertions(+), 24 deletions(-) create mode 100644 docs/reliability-merge-and-privacy-filter-plan.md create mode 100644 tests/test_jsonl_merge.py diff --git a/dataclaw/_cli/commands.py b/dataclaw/_cli/commands.py index 75c411e..4e469e6 100644 --- a/dataclaw/_cli/commands.py +++ b/dataclaw/_cli/commands.py @@ -239,9 +239,13 @@ def run_export( has_session_sources_fn: Callable[[str], bool], export_to_jsonl_fn: Callable[..., dict], summarize_jsonl_fn: Callable[[Path], dict], - push_to_huggingface_fn: Callable[[Path, str, dict], None], + push_to_huggingface_fn: Callable[..., None], ) -> None: config = load_config_fn() + redaction = { + "redact_strings": config.get("redact_strings", []) or [], + "redact_usernames": config.get("redact_usernames", []) or [], + } source_choice, source_explicit = _resolve_source_choice(args.source, config) source_filter = _normalize_source_filter(source_choice) @@ -420,7 +424,13 @@ def run_export( _print_export_elapsed(export_start_time) return - push_to_huggingface_fn(confirmed_file, repo_id, meta) + push_to_huggingface_fn(confirmed_file, repo_id, meta, redaction) + + # push_to_huggingface updates meta["sessions"] to the merged remote+local total; + # keep last_export.sessions consistent with what was actually published (H5). + last_export = config.get("last_export") + if isinstance(last_export, dict): + last_export["sessions"] = meta.get("sessions", last_export.get("sessions")) config["stage"] = "done" save_config_fn(config) @@ -596,7 +606,11 @@ def run_export( _print_export_elapsed(export_start_time) return - push_to_huggingface_fn(output_path, repo_id, meta) + push_to_huggingface_fn(output_path, repo_id, meta, redaction) + + last_export = config.get("last_export") + if isinstance(last_export, dict): + last_export["sessions"] = meta.get("sessions", last_export.get("sessions")) config["stage"] = "done" save_config_fn(config) diff --git a/dataclaw/_cli/exporting.py b/dataclaw/_cli/exporting.py index c4a3f08..aef0d50 100644 --- a/dataclaw/_cli/exporting.py +++ b/dataclaw/_cli/exporting.py @@ -17,6 +17,7 @@ from .. import _json as json from .._workers import configured_workers from ..anonymizer import Anonymizer +from ..jsonl_tools import merge_jsonl_union from ..parser import iter_project_sessions from ..providers import get_provider_non_anon_string_keys from ..secrets import transform_session @@ -648,8 +649,71 @@ def _build_breakdown_table(label: str, breakdown: object) -> str: return "\n".join(lines) -def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict) -> None: +REMOTE_CONVERSATIONS_FILE = "conversations.jsonl" +_MERGE_MAX_ATTEMPTS = 5 + + +def _build_carry_forward_redactor(redaction: dict | None): + """Build a ``redact_fn(record) -> record`` matching the export-time pipeline. + + Carried-forward remote records are re-redacted through the CURRENT redaction + policy (idempotent for already-current records) so old-policy redaction is + never republished. The Anonymizer + ``transform_session`` call are constructed + exactly the way the export loop builds them (``_export_to_jsonl_serial`` / + ``_export_session_task_worker``). + """ + redaction = redaction or {} + extra_usernames = list(redaction.get("redact_usernames") or []) + custom_strings = list(redaction.get("redact_strings") or []) + + def redact_fn(record: dict) -> dict: + # A fresh Anonymizer per record matches the per-worker construction and keeps + # pseudonyms deterministic (hash-based), so re-redaction stays idempotent. + anonymizer = Anonymizer(extra_usernames=extra_usernames) + source = record.get("source") or "" + redacted, _ = transform_session( + record, + anonymizer, + custom_strings=custom_strings, + non_anon_string_keys=get_provider_non_anon_string_keys(source), + ) + return redacted + + return redact_fn + + +def _download_remote_conversations(api, repo_id: str, dest: Path) -> tuple[Path | None, str | None]: + """Download the remote conversations file. Fail closed on anything but 404/no-repo. + + Returns ``(local_path, parent_commit_sha)``. ``local_path`` is ``None`` only when + the remote file (or repo) is confirmed absent, which means "empty remote". Any + other error (network/auth/HTTP) raises so the push aborts and the remote is never + overwritten with a local-only file. + """ + from huggingface_hub import hf_hub_download + from huggingface_hub.utils import EntryNotFoundError, RepositoryNotFoundError + + try: + parent_commit = api.repo_info(repo_id, repo_type="dataset").sha + except RepositoryNotFoundError: + return None, None + + try: + downloaded = hf_hub_download( + repo_id=repo_id, + filename=REMOTE_CONVERSATIONS_FILE, + repo_type="dataset", + revision=parent_commit, + local_dir=str(dest), + ) + except (EntryNotFoundError, RepositoryNotFoundError): + return None, parent_commit + return Path(downloaded), parent_commit + + +def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict, redaction: dict | None = None) -> None: from huggingface_hub import HfApi + from huggingface_hub.utils import HfHubHTTPError api = HfApi() @@ -663,17 +727,52 @@ def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict) -> None: ) print(f"Pushing to: {repo_id}") - try: - api.create_repo(repo_id, repo_type="dataset", exist_ok=True) + api.create_repo(repo_id, repo_type="dataset", exist_ok=True) - api.upload_file( - path_or_fileobj=str(jsonl_path), - path_in_repo="conversations.jsonl", - repo_id=repo_id, - repo_type="dataset", - commit_message="Update conversation data", - ) + redact_fn = _build_carry_forward_redactor(redaction) + + with tempfile.TemporaryDirectory(prefix="dataclaw-merge-") as temp_dir: + temp_path = Path(temp_dir) + merged_path = temp_path / "merged_conversations.jsonl" + + for attempt in range(1, _MERGE_MAX_ATTEMPTS + 1): + try: + remote_path, parent_commit = _download_remote_conversations(api, repo_id, temp_path / "remote") + except (OSError, ValueError, HfHubHTTPError) as e: + # Fail closed: never overwrite the remote with a local-only file. + emit_blocked_error( + f"reading remote dataset before merge (push aborted to avoid data loss): {e}" + ) + try: + merged_total = _merge_or_passthrough(remote_path, jsonl_path, merged_path, redact_fn) + except (OSError, ValueError) as e: + emit_blocked_error(f"merging remote and local conversations: {e}") + + # Keep metadata/README/shrink-gate consistent with what we actually upload. + meta["sessions"] = merged_total + upload_target = merged_path if remote_path is not None else jsonl_path + + try: + api.upload_file( + path_or_fileobj=str(upload_target), + path_in_repo=REMOTE_CONVERSATIONS_FILE, + repo_id=repo_id, + repo_type="dataset", + commit_message="Update conversation data (merged)", + parent_commit=parent_commit, + ) + break + except HfHubHTTPError as e: + status = getattr(getattr(e, "response", None), "status_code", None) + if status == 412 and attempt < _MERGE_MAX_ATTEMPTS: + print(f" Concurrent update detected (412), re-merging (attempt {attempt + 1})...") + continue + emit_blocked_error(f"uploading merged conversations to Hugging Face: {e}") + except (OSError, ValueError) as e: + emit_blocked_error(f"uploading merged conversations to Hugging Face: {e}") + + try: api.upload_file( path_or_fileobj=json.dumps_bytes(meta, indent=2), path_in_repo="metadata.json", @@ -689,13 +788,33 @@ def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict) -> None: repo_type="dataset", commit_message="Update dataset card", ) - except (OSError, ValueError) as e: + except (OSError, ValueError, HfHubHTTPError) as e: emit_blocked_error(f"uploading to Hugging Face: {e}") print(f"\nDataset: {hf_dataset_url(repo_id)}") print(f"Browse all: {hf_browse_tagged_url()}") +def _merge_or_passthrough(remote_path: Path | None, local_path: Path, merged_path: Path, redact_fn) -> int: + """Merge remote+local (or pass through local when remote is empty); return total.""" + if remote_path is None: + # No prior remote data: the local file is the merged set as-is. + with local_path.open("rb") as src: + total = sum(1 for line in src if line.strip()) + print(f"Merge: no prior remote dataset; publishing {total} local sessions") + return total + + stats = merge_jsonl_union(remote_path, local_path, merged_path, redact_fn=redact_fn) + print(stats.changelog_line()) + if stats.merged_total < stats.remote_total: + # Union invariant violated -> abort rather than risk a remote shrink. + emit_blocked_error( + f"merge produced fewer sessions ({stats.merged_total}) than the remote dataset " + f"({stats.remote_total}); aborting to prevent data loss." + ) + return stats.merged_total + + def _build_dataset_card(repo_id: str, meta: dict) -> str: sessions = meta.get("sessions", 0) model_breakdown = _normalize_breakdown(meta.get("model_breakdown"), normalize_key=_normalize_model_stats_key) diff --git a/dataclaw/jsonl_tools.py b/dataclaw/jsonl_tools.py index 219d904..af78434 100644 --- a/dataclaw/jsonl_tools.py +++ b/dataclaw/jsonl_tools.py @@ -135,6 +135,171 @@ class DiffResult: summary: dict[str, int] +@dataclass +class MergeStats: + """Outcome of a union merge of remote + local JSONL records.""" + + remote_total: int = 0 + local_total: int = 0 + merged_total: int = 0 + added: int = 0 # local-only records added + updated: int = 0 # records present in both where the local copy won (superset) + carried_forward: int = 0 # remote-only records preserved (re-redacted) + unchanged: int = 0 # records present in both where the remote copy won (re-redacted) + + def changelog_line(self) -> str: + return ( + f"Merge: added {self.added}, updated {self.updated}, " + f"carried_forward {self.carried_forward}, unchanged {self.unchanged} " + f"(remote {self.remote_total} -> merged {self.merged_total})" + ) + + +def merge_identity_key(obj: dict[str, Any]) -> tuple[Any, ...]: + """Dedup key for the union merge. + + Uses ``(source, session_id)`` when ``session_id`` is truthy, otherwise falls + back to the full ``identity_key()`` tuple. Keying on ``session_id`` avoids the + ``start_time`` format drift (H1) and anonymized ``project`` drift (H2) bugs that + would otherwise let one session appear under two keys (duplicate + leak). + """ + session_id = obj.get("session_id") + if session_id: + return ("sid", obj.get("source"), session_id) + return ("identity", *identity_key(obj)) + + +def _message_count(obj: dict[str, Any]) -> int: + messages = obj.get("messages") + return len(messages) if isinstance(messages, list) else 0 + + +def _end_time(obj: dict[str, Any]) -> str: + end_time = obj.get("end_time") + return end_time if isinstance(end_time, str) else "" + + +def _record_prefers(candidate: dict[str, Any], current: dict[str, Any]) -> bool: + """Return True if ``candidate`` should replace ``current`` in the union. + + Tie-break order: more messages, then larger canonical byte size, then later + ``end_time``. A tie on all three keeps ``current`` (caller decides which side + that is so "unchanged vs updated" classification stays meaningful). + """ + cand_messages = _message_count(candidate) + cur_messages = _message_count(current) + if cand_messages != cur_messages: + return cand_messages > cur_messages + + cand_bytes = len(canonical_record_bytes(candidate)) + cur_bytes = len(canonical_record_bytes(current)) + if cand_bytes != cur_bytes: + return cand_bytes > cur_bytes + + return _end_time(candidate) > _end_time(current) + + +def _load_raw_records(path: Path) -> list[dict[str, Any]]: + """Load JSONL records RAW (no diff normalization, preserves ``originalFile``).""" + records: list[dict[str, Any]] = [] + with path.open("rb") as handle: + for line in handle: + line = line.strip() + if not line: + continue + obj = orjson.loads(line) + if isinstance(obj, dict): + records.append(obj) + return records + + +def merge_jsonl_union( + remote_path: Path, + local_path: Path, + output_path: Path, + *, + redact_fn: Callable[[dict[str, Any]], dict[str, Any]], +) -> MergeStats: + """Union-merge remote + local JSONL into ``output_path``. + + Rules (keyed by :func:`merge_identity_key`): + - remote-only -> carry forward, re-redacted through ``redact_fn`` + - local-only -> add (already current-redacted upstream, passes through) + - both -> keep the superset (see :func:`_record_prefers`) + + Local records are written first when they win; carried-forward remote records + are always re-redacted via ``redact_fn`` before being written, which closes the + old-policy redaction-republish hole. The merge preserves first-seen ordering + (remote order, then any local-only additions) and guarantees + ``merged_total >= remote_total``. + + ``redact_fn`` is injected so this module stays free of import cycles with the + redaction pipeline. + """ + remote_records = _load_raw_records(remote_path) + local_records = _load_raw_records(local_path) + + stats = MergeStats(remote_total=len(remote_records), local_total=len(local_records)) + + # Build winning record per key, tracking origin (which side won) and which sides + # the key appeared on, so we can classify the change and re-redact correctly. + order: list[tuple[Any, ...]] = [] + winners: dict[tuple[Any, ...], dict[str, Any]] = {} + origin: dict[tuple[Any, ...], str] = {} # "remote" or "local" (winning side) + in_remote: set[tuple[Any, ...]] = set() + in_local: set[tuple[Any, ...]] = set() + + for record in remote_records: + key = merge_identity_key(record) + in_remote.add(key) + if key not in winners: + order.append(key) + winners[key] = record + origin[key] = "remote" + elif _record_prefers(record, winners[key]): + winners[key] = record + origin[key] = "remote" + + for record in local_records: + key = merge_identity_key(record) + in_local.add(key) + if key not in winners: + order.append(key) + winners[key] = record + origin[key] = "local" + elif _record_prefers(record, winners[key]): + winners[key] = record + origin[key] = "local" + + output_path.parent.mkdir(parents=True, exist_ok=True) + merged_total = 0 + with output_path.open("wb") as handle: + for key in order: + record = winners[key] + seen_remote = key in in_remote + seen_local = key in in_local + if origin[key] == "remote": + # Winning side is a remote record: re-redact through the CURRENT + # pipeline before publishing (closes the old-policy redaction hole). + record = redact_fn(record) + if seen_local: + stats.unchanged += 1 # in both, remote copy won + else: + stats.carried_forward += 1 # remote-only + else: + # Winning side is a local record (already current-redacted upstream). + if seen_remote: + stats.updated += 1 # in both, local copy won (superset) + else: + stats.added += 1 # local-only + handle.write(canonical_record_bytes(record)) + handle.write(b"\n") + merged_total += 1 + + stats.merged_total = merged_total + return stats + + def clean_strings(obj: Any) -> Any: if isinstance(obj, str): text = ANSI_RE.sub("", obj) diff --git a/docs/reliability-merge-and-privacy-filter-plan.md b/docs/reliability-merge-and-privacy-filter-plan.md new file mode 100644 index 0000000..6b4a1c7 --- /dev/null +++ b/docs/reliability-merge-and-privacy-filter-plan.md @@ -0,0 +1,161 @@ +# DataClaw reliability: merge-on-upload + privacy filter — design & plan + +> Produced from a multi-subagent audit (2026-05-30). Every claim is cited file:line. +> Two goals: (1) uploads must **merge**, never destroy, prior data; (2) the **model +> privacy filter** the UI promises must actually run. + +--- + +## Part A — Incremental / merge-on-upload + +### Current behavior (verified) +Upload is a **wholesale overwrite**. `push_to_huggingface` (`dataclaw/_cli/exporting.py:651`) +calls `api.upload_file(path_in_repo="conversations.jsonl", ...)` (`:669`) with the locally +re-generated full file. No download, no merge. The remote dataset always equals the latest +local scan, so any session not reproducible locally **right now** is silently deleted +(second machine, log rotation, narrower `--source`, new excluded project, cleared `~/.claude`). + +### Intent residue found +- `last_export_cutoff` — written by Rust (`app/src-tauri/src/dataclaw.rs:239-261`), **read by + nobody**. Half-built incremental-cutoff scaffold. +- `dataclaw/jsonl_tools.py` — has `IDENTITY_FIELDS=("source","project","session_id","start_time")` + (`:26`), `identity_key()` (`:221`), `index_jsonl()` (`:239`), `diff_jsonl_files()` (`:685`). + **No merge function; not wired to upload** — `diff-jsonl` is an offline CLI command only. +- Data-safety intent was actually shipped as **trust gates** (session-shrink `review.py:616`, + redaction-drift `commands.py:780`), all comparing to **local** `last_export` — not remote. + +### The correct unit +Each JSONL line is one **session** record with a `messages` array. "New messages since last +upload" = same `identity_key`, more messages, different record hash. The merge unit is the +**session**, not the message. + +### Recommended design: download → merge (union) → re-redact → re-gate → reupload + +1. **Download** prior remote `conversations.jsonl` via `hf_hub_download(repo_id, + "conversations.jsonl", repo_type="dataset")`. **Fail closed**: only a confirmed 404 + (`EntryNotFoundError`) or missing-repo (`RepositoryNotFoundError`) counts as "empty remote"; + any network/auth/HTTP error **aborts the push** (never overwrite remote with local-only). +2. **Union by `identity_key`** (read records RAW, not via `index_jsonl` — that normalizes and + strips `originalFile`, see H3): + - remote-only → carry forward (never drop) + - local-only → add + - both → keep superset (more messages; tie-break larger bytes, then later `end_time`) + - **Invariant `merged_total >= remote_total`** ⇒ shrink is impossible by construction. +3. **🔴 RE-REDACT carried-forward remote records through the CURRENT redactor** before writing + them (see "Critical correctness requirement" below). This is mandatory, not optional. +4. **Re-run the confirm gates on the MERGED file and re-hash it** so reviewed-bytes == + shipped-bytes (preserve the SHA invariant at `commands.py:365-376`). +5. **Upload the merged union** (not the local export) with `parent_commit=` + for optimistic concurrency; on 412 conflict re-download+re-merge (cap ~5 attempts). + Update `meta["sessions"]`/`last_export.sessions` to `merged_total` so metadata and the + shrink gate stay consistent (H5). + +Hook point: `push_to_huggingface` upload block (`exporting.py:666-675`). Merge fn belongs in +`jsonl_tools.py` next to `identity_key`/`index_jsonl`. Draft code exists in the agent transcript. + +### 🔴 Critical correctness requirement (the naive merge is a PRIVACY REGRESSION) +Remote records were redacted under **whatever policy was current when they were uploaded**. +Carrying them forward verbatim **re-publishes old, looser-policy redaction** and **bypasses +every gate** (gates run in `confirm()` on the local file, before the merge). Adversarial review +verdict on the naive union: scenarios (a) new redact_string, (b) superset-by-count picking the +older copy, (c) future model filter, (d) gate bypass → **leak, leak, leak, total-bypass**. + +**Fix:** re-run `secrets.transform_session` (+ `Anonymizer`) over every carried-forward record +with the **current** config before reupload. This is **idempotent and pseudonym-stable** +(verified): secrets become the constant `[REDACTED]` which matches no pattern; pseudonyms are +deterministic `user_+sha256(name)[:8]` and the hash token doesn't re-match the username regex. +Then re-gate + re-hash the merged file. + +### Additional holes the merge must handle +- **H1 `start_time` key drift** (`jsonl_tools.py:26`, `parsers/common.py:157-164`): format drift + (`Z` vs `+00:00`, `None` vs backfilled) → same session under two keys → **duplicate + leak**. + Fix: canonicalize `start_time` before keying, or key on `(source, project, session_id)`. +- **H2 anonymized `project` in key** (`secrets.py:572-576`): `project` is anonymized in-place and + is part of the key → key drift. Strip `project` from key or key on post-anon identity. +- **H3 `normalize_for_diff` strips `originalFile`** (`jsonl_tools.py:234-235`): don't read remote + records via the diff-normalizing loader or you destroy `originalFile` content. +- **H4 schema drift**: old remote records may lack fields the current walker expects → missed + secrets or crash. Re-redaction must be schema-tolerant. +- **H5 non-atomic upload + stale `last_export.sessions`**: 3 separate `upload_file` commits; + update session count to merged total. + +### Trust gates +Keep the existing **local** gates unchanged (they guard the human confirm step). Do **not** add a +remote-shrink gate — the union invariant makes remote shrink impossible. But the merged file +**must** pass the gates after merge+re-redaction (step 4). + +--- + +## Part B — Model-based privacy filter + +### It was real, and was lost (not reverted) +A complete **848-line `dataclaw/privacy_filter.py`** + `tests/test_privacy_filter.py` was +committed on `5d0a741` ("Add macOS app release flow", a side branch) and **never merged to main**. +A *different* same-titled commit (`42b21d1`) landed on main without it. The file was on disk as +late as the 2026-05-16 desloppify scan, then lost untracked. The `pii` deps +(`transformers`, `torch`, `accelerate`, `tokenizers`) were dropped from HEAD `pyproject.toml`. + +Recover with: `git show 5d0a741:dataclaw/privacy_filter.py` and `:tests/test_privacy_filter.py`. + +### What it does (and quality) +`transformers.pipeline(task="token-classification", aggregation_strategy="simple")` NER with real +CPU/MPS device handling (`resolve_device`/`_auto_device`), 480-token chunking, min_score 0.85, +oversized-session blanket-redact guard, reverse-order span splicing. **Core logic is sound.** + +Bugs / gaps (line refs on the `5d0a741` version): +- 🔴 **Field-walk mismatch (must fix):** it walks `messages[].content/thinking` + `tool_uses[].input/output` + but **NOT `messages[].content_parts`**, which HEAD's `secrets.transform_session` *does* walk + (`secrets.py:590`). PII in `content_parts` would pass untouched. Add `content_parts` to the + field loops in `scan_session`/`redact_session`/`_redact_oversized_session` + a test. +- 🔴 **`MODEL_ID = "openai/privacy-filter"` is a placeholder** — does not exist on the Hub. +- Minor: cross-chunk-boundary entities may be split/under-redacted (low severity); `min_score` + discarded in `_load` then re-applied in `scan_text` (cosmetic); `dtype=` kwarg needs transformers ≥4.45. + +### Model recommendation (drop-in for the existing pipeline call) +- **Default: `iiiorg/piiranha-v1-detect-personal-information`** — MIT, ~280 MB, DeBERTa + token-classification, MPS-ok, drops in with **zero adapter code**. +- **Fallback: `lakshyakh93/deberta_finetuned_pii`** — broader labels, also a pure drop-in. +- Avoid GLiNER / Presidio for v1 (need adapters / heavier deps). Verify Hub availability and that + `aggregation_strategy="simple"` yields `start`/`end` offsets before committing. + +### Rewiring against HEAD `_cli` structure +- **Mutation happens only at EXPORT time**, inline per session, right after + `secrets.transform_session` (`exporting.py:201` parallel worker, `:292` serial). Confirm + (`review.py:625`) is **read-only** and hashes the file (`review.py:830`); publish enforces that + hash (`commands.py:365-376`). **So the model edits MUST run at export, before the hash lock** — + not at confirm. +- Keep the dict/text functions (`redact_session`, `redact_text`, `scan_text`, `_load`, + `resolve_device`, oversized guard); **discard** the shard/jsonl/manifest layer (HEAD has no + run_dir/manifest pipeline — it was dropped). +- Read config (free-form dict, no schema in `config.py`): + `privacy_filter.{enabled (default False), device, min_score=0.85, include_tool_io=True, roles={user}}`. + Rust already writes `privacy_filter.enabled/.device` (`dataclaw.rs:571,580`) — this closes the gap. +- Emit `privacy_filter_*` events via `progress_callback` → lights up the **already-present** dead + Dashboard `model_privacy` stage (`Dashboard.tsx:209-210`, handlers `:434-516`). `mechanical_pii` + stage maps to the existing regex layer. + +### Deps / packaging +- Restore `[project.optional-dependencies] pii = ["transformers>=4.57","torch>=2.3","accelerate","tokenizers"]` + + the `pii` pytest marker. Keep as an **extra** so the lazy-import graceful-degradation holds. +- **Do NOT bundle torch into the Mac sidecar by default** (~1.5–2.5 GB; `pyinstaller.spec` has no + torch hooks). Default `enabled=false`. Treat the model stage as power-user / `dataclaw[pii]`, or + a separate heavy build. **Download the model at first run** (HF cache), don't bundle weights. +- **Graceful degradation:** if enabled but torch/model/network unavailable → **warn and continue + with mechanical-only redaction**, do not abort the export. + +--- + +## Prioritized action list +1. **Merge-on-upload with re-redaction + re-gate** (Part A). The data-destruction bug *and* the + privacy regression it would introduce — do them together; one is unsafe without the other. +2. **Atomic export write** (temp+fsync+rename in `exporting.py:493`) — independent, stops silently + truncated datasets being published. (From the first audit.) +3. **Privacy filter recovery + content_parts fix + real MODEL_ID + export-time rewire** (Part B). +4. **Source enum reconcile** (drop `hermes`, add `cursor`, validate in Rust, fail-closed in Python) + and **harden secret patterns + salt the anonymizer**. (From the first audit.) +5. Real cross-process run lock; batch the 3 HF uploads into one `create_commit`. + +## Open verification items (couldn't run live) +- `huggingface_hub` version: confirm `upload_file(parent_commit=...)` support and that conflicts + surface as HTTP 412; confirm `hf_hub_download` error classes. +- Live Hub availability + offset behavior of the two recommended `MODEL_ID`s. diff --git a/tests/test_cli_exporting.py b/tests/test_cli_exporting.py index 9059fc9..2ae5291 100644 --- a/tests/test_cli_exporting.py +++ b/tests/test_cli_exporting.py @@ -1,6 +1,7 @@ """Tests for CLI export and publish helpers.""" from concurrent.futures import Future +from pathlib import Path from unittest.mock import MagicMock, patch import pytest @@ -597,21 +598,118 @@ def test_writes_multi_mb_blob_verbatim(self, tmp_path, mock_anonymizer): assert rows[0]["messages"][0]["content_parts"][0]["source"]["data"] == blob +class _FakeEntryNotFoundError(Exception): + pass + + +class _FakeRepositoryNotFoundError(Exception): + pass + + +class _FakeHfHubHTTPError(Exception): + def __init__(self, message="", status_code=None): + super().__init__(message) + self.response = MagicMock(status_code=status_code) + + +def _install_hf_mocks(mock_api, *, downloaded_path=None, download_error=None): + """Build mock ``huggingface_hub`` + ``huggingface_hub.utils`` modules. + + ``hf_hub_download`` returns ``downloaded_path`` (or raises ``download_error``). + Returns the list/dict of submodules to feed ``patch.dict("sys.modules", ...)``. + """ + + def fake_download(*args, **kwargs): + if download_error is not None: + raise download_error + return str(downloaded_path) + + utils_mod = MagicMock( + EntryNotFoundError=_FakeEntryNotFoundError, + RepositoryNotFoundError=_FakeRepositoryNotFoundError, + HfHubHTTPError=_FakeHfHubHTTPError, + ) + hf_mod = MagicMock( + HfApi=MagicMock(return_value=mock_api), + hf_hub_download=fake_download, + ) + hf_mod.utils = utils_mod + return {"huggingface_hub": hf_mod, "huggingface_hub.utils": utils_mod} + + class TestPushToHuggingface: - def test_success_flow(self, tmp_path): - jsonl_path = tmp_path / "data.jsonl" - jsonl_path.write_text("{}\n") + def test_merges_remote_then_uploads_merged(self, tmp_path): + # Remote has one session; local has a different session -> union of both. + remote = tmp_path / "remote.jsonl" + remote.write_text('{"source":"claude","session_id":"r1","messages":[{"role":"user"}]}\n') + local = tmp_path / "local.jsonl" + local.write_text('{"source":"claude","session_id":"l1","messages":[{"role":"user"}]}\n') + + captured = {} mock_api = MagicMock() mock_api.whoami.return_value = {"name": "alice"} - mock_hfapi_cls = MagicMock(return_value=mock_api) + mock_api.repo_info.return_value = MagicMock(sha="abc123") - with patch.dict("sys.modules", {"huggingface_hub": MagicMock(HfApi=mock_hfapi_cls)}): - push_to_huggingface(jsonl_path, "user/repo", {}) + def fake_upload(*args, **kwargs): + if kwargs.get("path_in_repo") == "conversations.jsonl": + captured["uploaded"] = Path(kwargs["path_or_fileobj"]).read_text() + captured["parent_commit"] = kwargs.get("parent_commit") - mock_api.create_repo.assert_called_once_with("user/repo", repo_type="dataset", exist_ok=True) + mock_api.upload_file.side_effect = fake_upload + + with patch.dict("sys.modules", _install_hf_mocks(mock_api, downloaded_path=remote)): + push_to_huggingface(local, "user/repo", {}, {"redact_strings": [], "redact_usernames": []}) + + # Both remote-only and local-only sessions present in the uploaded merge. + assert '"r1"' in captured["uploaded"] + assert '"l1"' in captured["uploaded"] + assert captured["parent_commit"] == "abc123" assert mock_api.upload_file.call_count == 3 + def test_empty_remote_passthrough(self, tmp_path): + local = tmp_path / "local.jsonl" + local.write_text('{"source":"claude","session_id":"l1","messages":[{"role":"user"}]}\n') + + mock_api = MagicMock() + mock_api.whoami.return_value = {"name": "alice"} + mock_api.repo_info.return_value = MagicMock(sha="abc123") + + captured = {} + + def fake_upload(*args, **kwargs): + if kwargs.get("path_in_repo") == "conversations.jsonl": + captured["uploaded"] = Path(kwargs["path_or_fileobj"]).read_text() + + mock_api.upload_file.side_effect = fake_upload + + mocks = _install_hf_mocks(mock_api, download_error=_FakeEntryNotFoundError("404")) + with patch.dict("sys.modules", mocks): + push_to_huggingface(local, "user/repo", {}, {"redact_strings": [], "redact_usernames": []}) + + assert '"l1"' in captured["uploaded"] + assert mock_api.upload_file.call_count == 3 + + def test_fail_closed_on_download_error(self, tmp_path): + from dataclaw._cli.common import CLIBlockedError + + local = tmp_path / "local.jsonl" + local.write_text('{"source":"claude","session_id":"l1","messages":[{"role":"user"}]}\n') + + mock_api = MagicMock() + mock_api.whoami.return_value = {"name": "alice"} + mock_api.repo_info.return_value = MagicMock(sha="abc123") + + # A non-404 HTTP error during download must abort the push (never upload local-only). + mocks = _install_hf_mocks(mock_api, download_error=_FakeHfHubHTTPError("500 server error")) + with patch.dict("sys.modules", mocks): + with pytest.raises(CLIBlockedError): + push_to_huggingface(local, "user/repo", {}, {"redact_strings": [], "redact_usernames": []}) + + # conversations.jsonl must NOT have been uploaded. + uploaded_paths = [c.kwargs.get("path_in_repo") for c in mock_api.upload_file.call_args_list] + assert "conversations.jsonl" not in uploaded_paths + def test_auth_failure(self, tmp_path): from dataclaw._cli.common import CLIBlockedError @@ -620,9 +718,8 @@ def test_auth_failure(self, tmp_path): mock_api = MagicMock() mock_api.whoami.side_effect = OSError("Auth failed") - mock_hf_module = MagicMock(HfApi=MagicMock(return_value=mock_api)) - with patch.dict("sys.modules", {"huggingface_hub": mock_hf_module}): + with patch.dict("sys.modules", _install_hf_mocks(mock_api)): with pytest.raises(CLIBlockedError): push_to_huggingface(jsonl_path, "user/repo", {}) diff --git a/tests/test_cli_facade.py b/tests/test_cli_facade.py index ce80892..1690419 100644 --- a/tests/test_cli_facade.py +++ b/tests/test_cli_facade.py @@ -223,10 +223,11 @@ def test_push_reuses_confirmed_file(self, tmp_path, monkeypatch, capsys): lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not regenerate jsonl")), ) - def fake_push(path, repo_id, meta): + def fake_push(path, repo_id, meta, redaction=None): pushed["path"] = path pushed["repo_id"] = repo_id pushed["meta"] = meta + pushed["redaction"] = redaction monkeypatch.setattr("dataclaw.cli.push_to_huggingface", fake_push) monkeypatch.setattr( diff --git a/tests/test_jsonl_merge.py b/tests/test_jsonl_merge.py new file mode 100644 index 0000000..e685230 --- /dev/null +++ b/tests/test_jsonl_merge.py @@ -0,0 +1,218 @@ +"""Tests for the union merge-on-upload helper (Part A).""" + +import orjson + +from dataclaw import jsonl_tools +from dataclaw.jsonl_tools import MergeStats, merge_identity_key, merge_jsonl_union + + +def _write_jsonl(path, records): + with path.open("wb") as handle: + for record in records: + handle.write(orjson.dumps(record)) + handle.write(b"\n") + + +def _read_jsonl(path): + return [orjson.loads(line) for line in path.read_bytes().splitlines() if line.strip()] + + +def _identity(record): + """A no-op redact_fn used when re-redaction content does not matter.""" + return record + + +class TestMergeIdentityKey: + def test_keys_on_source_and_session_id_when_present(self): + record = {"source": "claude", "session_id": "s1", "project": "p", "start_time": "t"} + assert merge_identity_key(record) == ("sid", "claude", "s1") + + def test_falls_back_to_full_identity_when_no_session_id(self): + record = {"source": "claude", "project": "p", "start_time": "t"} + key = merge_identity_key(record) + assert key[0] == "identity" + # full identity_key tuple follows + assert key[1:] == jsonl_tools.identity_key(record) + + +class TestMergeJsonlUnion: + def test_remote_only_carried_forward(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl(remote, [{"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}]) + _write_jsonl(local, []) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert [r["session_id"] for r in merged] == ["r1"] + assert stats.carried_forward == 1 + assert stats.added == 0 + assert stats.merged_total == 1 >= stats.remote_total + + def test_local_only_added(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl(remote, []) + _write_jsonl(local, [{"source": "claude", "session_id": "l1", "messages": [{"role": "user"}]}]) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert [r["session_id"] for r in merged] == ["l1"] + assert stats.added == 1 + assert stats.carried_forward == 0 + + def test_both_superset_picks_more_messages(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [{"source": "claude", "session_id": "s1", "messages": [{"role": "user"}]}], + ) + _write_jsonl( + local, + [{"source": "claude", "session_id": "s1", "messages": [{"role": "user"}, {"role": "assistant"}]}], + ) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert len(merged) == 1 + assert len(merged[0]["messages"]) == 2 # local superset won + assert stats.updated == 1 + assert stats.merged_total == 1 >= stats.remote_total + + def test_both_remote_has_more_keeps_remote(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [{"source": "claude", "session_id": "s1", "messages": [{"role": "user"}, {"role": "assistant"}]}], + ) + _write_jsonl( + local, + [{"source": "claude", "session_id": "s1", "messages": [{"role": "user"}]}], + ) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert len(merged) == 1 + assert len(merged[0]["messages"]) == 2 # remote superset kept + assert stats.unchanged == 1 + assert stats.updated == 0 + + def test_dedup_by_source_session_id_ignores_start_time_and_project_drift(self, tmp_path): + # H1/H2: same (source, session_id) but drifted start_time + project. + # Must NOT duplicate. + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [ + { + "source": "claude", + "session_id": "s1", + "project": "old-anon-name", + "start_time": "2025-01-01T00:00:00Z", + "messages": [{"role": "user"}], + } + ], + ) + _write_jsonl( + local, + [ + { + "source": "claude", + "session_id": "s1", + "project": "new-anon-name", + "start_time": "2025-01-01T00:00:00+00:00", + "messages": [{"role": "user"}], + } + ], + ) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert len(merged) == 1 # collapsed to one despite drift + assert stats.merged_total == 1 + assert stats.merged_total >= stats.remote_total + + def test_original_file_preserved_on_carried_forward(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [ + { + "source": "claude", + "session_id": "r1", + "messages": [{"role": "user", "originalFile": "secret-original-content"}], + } + ], + ) + _write_jsonl(local, []) + + merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = _read_jsonl(out) + assert merged[0]["messages"][0]["originalFile"] == "secret-original-content" + + def test_reredaction_applied_to_carried_forward(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl(remote, [{"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}]) + _write_jsonl(local, [{"source": "claude", "session_id": "l1", "messages": [{"role": "user"}]}]) + + called_on = [] + + def fake_redact(record): + called_on.append(record.get("session_id")) + record["reredacted"] = True + return record + + merge_jsonl_union(remote, local, out, redact_fn=fake_redact) + + # redact_fn must be called on the carried-forward remote record (r1) but not l1. + assert called_on == ["r1"] + merged = {r["session_id"]: r for r in _read_jsonl(out)} + assert merged["r1"].get("reredacted") is True + assert "reredacted" not in merged["l1"] + + def test_union_invariant_merged_ge_remote(self, tmp_path): + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [ + {"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}, + {"source": "claude", "session_id": "r2", "messages": [{"role": "user"}]}, + ], + ) + # Local re-exports only r1 (e.g. narrower --source). r2 must survive. + _write_jsonl(local, [{"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}]) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + merged = {r["session_id"] for r in _read_jsonl(out)} + assert merged == {"r1", "r2"} + assert stats.merged_total >= stats.remote_total + + def test_changelog_line_format(self): + stats = MergeStats( + remote_total=3, local_total=2, merged_total=4, added=1, updated=1, carried_forward=2, unchanged=0 + ) + line = stats.changelog_line() + assert "added 1" in line + assert "carried_forward 2" in line + assert "remote 3 -> merged 4" in line From ffece913c3e8467ba6094733c5e25aaf2283615b Mon Sep 17 00:00:00 2001 From: POM Date: Sat, 30 May 2026 02:05:46 +0200 Subject: [PATCH 2/5] Restore model privacy filter and wire it into export Recover the never-merged dataclaw/privacy_filter.py (token-classification PII NER) and make the UI "Privacy filter" toggle real: - Default model openai/privacy-filter (Apache-2.0, the originally-intended model), overridable via privacy_filter.model config / env var. - Fix the field-walk gap: scan/redact messages[].content_parts too (HEAD's transform_session walks it; the recovered code did not). - Wire the model filter into the export loop (serial + parallel worker), applied in-memory right after transform_session, so edits are baked into the file confirm hashes and publish enforces. Reads privacy_filter.{enabled(default false),device,min_score,model} from config. - Graceful degradation: if torch/transformers/model is unavailable, warn once and continue with mechanical redaction only -- never abort the export. - Lazy-import privacy_filter so the CLI never pulls in torch unless enabled. - Drop the dead shard/manifest layer; keep dict/text functions + oversized guard. Restore the pii optional-dependency extra + pytest marker. torch stays out of the Mac sidecar by default (enabled=false); the model downloads on first use. Co-Authored-By: Claude Opus 4.8 (1M context) --- dataclaw/_cli/exporting.py | 99 ++- dataclaw/privacy_filter.py | 637 ++++++++++++++++++ ...liability-merge-and-privacy-filter-plan.md | 17 +- pyproject.toml | 8 + tests/conftest.py | 13 + tests/test_privacy_filter.py | 382 +++++++++++ 6 files changed, 1148 insertions(+), 8 deletions(-) create mode 100644 dataclaw/privacy_filter.py create mode 100644 tests/test_privacy_filter.py diff --git a/dataclaw/_cli/exporting.py b/dataclaw/_cli/exporting.py index aef0d50..383caca 100644 --- a/dataclaw/_cli/exporting.py +++ b/dataclaw/_cli/exporting.py @@ -155,6 +155,88 @@ class _WorkerSessionResult: skipped_model: bool = False +@dataclass(frozen=True, slots=True) +class _PrivacyFilterConfig: + """Resolved model privacy-filter settings. + + Read once at export time and passed by value into the parallel worker + process (which has no access to the parent's loaded config), mirroring how + ``custom_strings``/``extra_usernames`` are already threaded through. + """ + + enabled: bool = False + device: str | None = None + min_score: float = 0.85 + model: str | None = None + + +def _read_privacy_filter_config() -> _PrivacyFilterConfig: + """Read the free-form ``privacy_filter.*`` config block (no schema). + + Defaults: ``enabled=False``, ``device=auto`` (resolved lazily inside the + filter), ``min_score=0.85``, ``model=None`` (filter falls back to its + default PII model id / env override). + """ + try: + from ..config import load_config + cfg = load_config() + except Exception: # pragma: no cover - config IO is best-effort + return _PrivacyFilterConfig() + pf_cfg = cfg.get("privacy_filter") if isinstance(cfg, dict) else None + if not isinstance(pf_cfg, dict): + return _PrivacyFilterConfig() + + enabled = bool(pf_cfg.get("enabled", False)) + device = pf_cfg.get("device") + device = device.strip() if isinstance(device, str) and device.strip() else None + min_score = pf_cfg.get("min_score", 0.85) + try: + min_score = float(min_score) + except (TypeError, ValueError): + min_score = 0.85 + model = pf_cfg.get("model") + model = model.strip() if isinstance(model, str) and model.strip() else None + return _PrivacyFilterConfig(enabled=enabled, device=device, min_score=min_score, model=model) + + +_PF_WARNED = False + + +def _apply_model_privacy_filter(session: dict, pf_config: _PrivacyFilterConfig) -> dict: + """Run the optional model PII filter over an already-redacted session. + + Graceful degradation: this runs AFTER ``secrets.transform_session`` (the + mechanical redaction has already happened and is preserved). If the model + stack (torch/transformers/the model download) is unavailable for any + reason, warn once and return the session unchanged — the model stage is + additive safety, never a hard dependency for a successful export. + """ + global _PF_WARNED + if not pf_config.enabled: + return session + try: + # Lazy import so importing this module / the CLI never pulls in torch. + from .. import privacy_filter + if not privacy_filter.is_available(): + raise RuntimeError("torch/transformers not installed") + redacted, _findings = privacy_filter.redact_session( + session, + device=pf_config.device, + min_score=pf_config.min_score, + model=pf_config.model, + ) + return redacted + except Exception as exc: # noqa: BLE001 - never let the model stage abort export + if not _PF_WARNED: + _PF_WARNED = True + print( + f" Warning: model privacy filter skipped ({exc}); " + "continuing with mechanical redaction only.", + file=sys.stderr, + ) + return session + + def _export_extra_usernames(anonymizer: Anonymizer) -> tuple[str, ...]: extra = getattr(anonymizer, "_extra_dict", {}) if isinstance(extra, dict): @@ -180,7 +262,7 @@ def _can_parallelize_export(parse_project_sessions_fn, task_count: int, workers: def _export_session_task_worker(payload) -> _WorkerSessionResult: - task, include_thinking, custom_strings, extra_usernames = payload + task, include_thinking, custom_strings, extra_usernames, pf_config = payload anonymizer = Anonymizer(extra_usernames=list(extra_usernames)) try: session = parse_export_session_task(task, anonymizer, include_thinking) @@ -205,6 +287,8 @@ def _export_session_task_worker(payload) -> _WorkerSessionResult: custom_strings=custom_strings, non_anon_string_keys=get_provider_non_anon_string_keys(task.source), ) + # Optional model privacy filter runs AFTER mechanical redaction, before hashing. + session = _apply_model_privacy_filter(session, pf_config) fingerprint = _gemini_dedupe_fingerprint(session, task.source) stats = session.get("stats", {}) input_tokens, output_tokens = _token_totals(stats) @@ -260,6 +344,7 @@ def _export_to_jsonl_serial( default_source: str, include_thinking: bool, custom_strings: list[str] | None, + pf_config: _PrivacyFilterConfig, ) -> dict: total = 0 skipped = 0 @@ -296,6 +381,8 @@ def _export_to_jsonl_serial( custom_strings=custom_strings, non_anon_string_keys=get_provider_non_anon_string_keys(source), ) + # Optional model privacy filter runs AFTER mechanical redaction, before hashing. + session = _apply_model_privacy_filter(session, pf_config) total_redactions += n_redacted fingerprint = _gemini_dedupe_fingerprint(session, source) @@ -358,6 +445,7 @@ def _export_to_jsonl_parallel( tasks: list[ExportSessionTask], workers: int, anonymizer: Anonymizer, + pf_config: _PrivacyFilterConfig, ) -> dict: total = 0 skipped = 0 @@ -398,7 +486,7 @@ def submit_ready(executor: ProcessPoolExecutor) -> None: state = project_state[task.project_index] if state["start_time"] is None: state["start_time"] = time.perf_counter() - payload = (task, include_thinking, custom_strings, extra_usernames) + payload = (task, include_thinking, custom_strings, extra_usernames, pf_config) future = executor.submit(_export_session_task_worker, payload) pending[future] = order_index @@ -495,6 +583,11 @@ def export_to_jsonl( except OSError as e: emit_blocked_error(f"cannot write to {output_path}: {e}") + pf_config = _read_privacy_filter_config() + if pf_config.enabled: + # stderr so it doesn't interleave with the per-project stdout summaries. + print(" Model privacy filter enabled; will run after mechanical redaction.", file=sys.stderr) + with fh as f: if parse_project_sessions_fn is iter_project_sessions: tasks = build_export_session_tasks(selected_projects, default_source) @@ -508,6 +601,7 @@ def export_to_jsonl( tasks, resolved_workers, anonymizer, + pf_config, ) return _export_to_jsonl_serial( selected_projects, @@ -517,6 +611,7 @@ def export_to_jsonl( default_source, include_thinking, custom_strings, + pf_config, ) diff --git a/dataclaw/privacy_filter.py b/dataclaw/privacy_filter.py new file mode 100644 index 0000000..9906dd6 --- /dev/null +++ b/dataclaw/privacy_filter.py @@ -0,0 +1,637 @@ +"""Optional privacy-filter PII scanning with lazy model imports.""" + +import hashlib +import json +import logging +import re +from collections.abc import Callable +from dataclasses import dataclass, replace +from typing import Any, Iterable + +# Default model: OpenAI's Apache-2.0 licensed PII detector, a +# token-classification model that drops into the existing +# ``pipeline(task="token-classification", aggregation_strategy="simple")`` call +# with no adapter code. Overridable via the ``privacy_filter.model`` config key +# (or the ``DATACLAW_PRIVACY_FILTER_MODEL`` env var) so it is not hard-locked. +DEFAULT_MODEL_ID = "openai/privacy-filter" +MODEL_ID = DEFAULT_MODEL_ID +_DEFAULT_MIN_SCORE = 0.85 +_CHUNK_TOKENS = 480 +_TEXT_PROGRESS_MIN_CHARS = 20_000 +_TEXT_PROGRESS_CHUNK_INTERVAL = 25 +_MAX_MODEL_SESSION_CHARS = 25_000_000 +_MAX_MODEL_SESSION_STRINGS = 40_000 + +_LOG = logging.getLogger(__name__) +_PIPELINES: dict[tuple[str, str | None, str], Any] = {} +_TOKEN_RE = re.compile(r"\S+") +ProgressCallback = Callable[[str, dict[str, Any]], None] + +# Mapping from string dtype config values to torch dtypes. Resolved lazily so +# that importing this module never imports torch. +_DTYPE_ALIASES = {"auto", "fp32", "float32", "bf16", "bfloat16", "fp16", "float16"} + + +@dataclass(frozen=True) +class Finding: + entity: str + text: str + score: float + start: int | None = None + end: int | None = None + field: str | None = None + session_id: str | None = None + source: str | None = None + + def fingerprint(self) -> str: + return hashlib.sha256(f"{self.entity}|{self.text}".encode()).hexdigest() + + +def is_available() -> bool: + try: + import torch # noqa: F401 + import transformers # noqa: F401 + except Exception: + return False + return True + + +def _config_dtype() -> str: + """Read ``config.privacy_filter.dtype`` lazily; default to ``"auto"``.""" + try: + from . import config as _config_mod + cfg = _config_mod.load_config() + except Exception: # pragma: no cover - config IO is best-effort + return "auto" + pf_cfg = cfg.get("privacy_filter") if isinstance(cfg, dict) else None + if isinstance(pf_cfg, dict): + value = pf_cfg.get("dtype") + if isinstance(value, str) and value.lower() in _DTYPE_ALIASES: + return value.lower() + return "auto" + + +def _config_model() -> str: + """Read ``config.privacy_filter.model`` lazily; default to ``MODEL_ID``. + + Env var ``DATACLAW_PRIVACY_FILTER_MODEL`` takes precedence so the model is + never hard-locked even when no config file is present. + """ + import os + + env_value = os.environ.get("DATACLAW_PRIVACY_FILTER_MODEL") + if isinstance(env_value, str) and env_value.strip(): + return env_value.strip() + try: + from . import config as _config_mod + cfg = _config_mod.load_config() + except Exception: # pragma: no cover - config IO is best-effort + return MODEL_ID + pf_cfg = cfg.get("privacy_filter") if isinstance(cfg, dict) else None + if isinstance(pf_cfg, dict): + value = pf_cfg.get("model") + if isinstance(value, str) and value.strip(): + return value.strip() + return MODEL_ID + + +def _config_device() -> str | None: + """Read ``config.privacy_filter.device`` lazily.""" + try: + from . import config as _config_mod + cfg = _config_mod.load_config() + except Exception: # pragma: no cover - config IO is best-effort + return None + pf_cfg = cfg.get("privacy_filter") if isinstance(cfg, dict) else None + if isinstance(pf_cfg, dict): + value = pf_cfg.get("device") + if isinstance(value, str) and value.strip(): + return value.strip() + return None + + +def _auto_device() -> str: + try: + import torch + + if getattr(torch.backends, "mps", None) is not None and torch.backends.mps.is_available(): + return "mps" + except Exception: + pass + return "cpu" + + +def resolve_device(device: str | None = None) -> str: + value = device if device is not None else _config_device() + if not isinstance(value, str) or not value.strip() or value.strip().lower() == "auto": + return _auto_device() + return value.strip() + + +def _resolve_dtype(name: str, device: str | None = None) -> Any: + """Translate a dtype name to a ``torch.dtype``, applying ``auto`` rules. + + - ``auto`` -> ``bfloat16`` only when the configured device is MPS, else + ``float32``. CPU is the default for predictable unattended runs. + - ``fp32``/``float32`` -> ``torch.float32`` + - ``bf16``/``bfloat16`` -> ``torch.bfloat16`` + - ``fp16``/``float16`` -> ``torch.float16`` + """ + import torch + + key = (name or "auto").lower() + if key == "auto": + return torch.bfloat16 if str(device or "").lower().startswith("mps") else torch.float32 + if key in ("bf16", "bfloat16"): + return torch.bfloat16 + if key in ("fp16", "float16"): + return torch.float16 + return torch.float32 + + +def _build_pipeline(**kwargs: Any) -> Any: + """Indirection so tests can patch pipeline construction without monkey- + patching the (lazy) ``transformers`` module itself.""" + from transformers import pipeline + return pipeline(**kwargs) + + +def _load( + *, + device: str | None = None, + min_score: float = _DEFAULT_MIN_SCORE, + dtype: str | None = None, + model: str | None = None, +) -> Any: + del min_score + effective_device = resolve_device(device) + dtype_name = (dtype or _config_dtype()).lower() + effective_model = model if (isinstance(model, str) and model.strip()) else _config_model() + key = (effective_model, effective_device, dtype_name) + if key not in _PIPELINES: + torch_dtype = _resolve_dtype(dtype_name, effective_device) + _LOG.info( + "privacy_filter_model_load_started", + extra={ + "phase": "privacy_filter", + "extra": {"model": effective_model, "device": effective_device, "dtype": dtype_name}, + }, + ) + # transformers >=4.45 deprecated ``torch_dtype`` in favour of ``dtype``; + # we pin >=4.57 in pyproject.toml so the new name is always available. + kwargs: dict[str, Any] = { + "task": "token-classification", + "model": effective_model, + "aggregation_strategy": "simple", + "dtype": torch_dtype, + } + if effective_device is not None: + kwargs["device"] = effective_device + _PIPELINES[key] = _build_pipeline(**kwargs) + _LOG.info( + "privacy_filter_model_load_finished", + extra={ + "phase": "privacy_filter", + "extra": {"model": effective_model, "device": effective_device, "dtype": dtype_name}, + }, + ) + return _PIPELINES[key] + + +def _chunk_by_tokens(text: str, max_tokens: int = _CHUNK_TOKENS) -> Iterable[tuple[str, int]]: + group: list[re.Match[str]] = [] + saw_match = False + for match in _TOKEN_RE.finditer(text): + saw_match = True + group.append(match) + if len(group) >= max_tokens: + yield text[group[0].start():group[-1].end()], group[0].start() + group.clear() + if group: + yield text[group[0].start():group[-1].end()], group[0].start() + if not saw_match and text: + yield text, 0 + + +def _entity_name(raw: dict[str, Any]) -> str: + value = raw.get("entity_group") or raw.get("entity") or raw.get("label") + return str(value or "PII") + + +def _entity_text(raw: dict[str, Any], chunk: str) -> str: + word = raw.get("word") or raw.get("text") + if isinstance(word, str) and word.strip(): + return word.strip() + start = raw.get("start") + end = raw.get("end") + if isinstance(start, int) and isinstance(end, int): + return chunk[start:end] + return "" + + +def scan_text( + text: str, + *, + device: str | None = None, + min_score: float = _DEFAULT_MIN_SCORE, + progress_callback: ProgressCallback | None = None, + field: str | None = None, + session_id: str | None = None, + source: str | None = None, + model: str | None = None, +) -> list[Finding]: + pipe = _load(device=device, min_score=min_score, model=model) + findings: list[Finding] = [] + should_log = progress_callback is not None and len(text) >= _TEXT_PROGRESS_MIN_CHARS + if should_log: + progress_callback("privacy_filter_text_started", { + "field": field, + "session_id": session_id, + "source": source, + "char_count": len(text), + "chunk_count": None, + }) + chunk_index = 0 + for chunk_index, (chunk, offset) in enumerate(_chunk_by_tokens(text), start=1): + for raw in pipe(chunk): + if not isinstance(raw, dict): + continue + score = float(raw.get("score") or 0.0) + if score < min_score: + continue + start = raw.get("start") + end = raw.get("end") + findings.append(Finding( + entity=_entity_name(raw), + text=_entity_text(raw, chunk), + score=score, + start=offset + start if isinstance(start, int) else None, + end=offset + end if isinstance(end, int) else None, + )) + if should_log and ( + chunk_index == 1 + or chunk_index % _TEXT_PROGRESS_CHUNK_INTERVAL == 0 + ): + progress_callback("privacy_filter_text_progress", { + "field": field, + "session_id": session_id, + "source": source, + "char_count": len(text), + "chunk_index": chunk_index, + "chunk_count": None, + "findings": len(findings), + }) + if should_log: + progress_callback("privacy_filter_text_finished", { + "field": field, + "session_id": session_id, + "source": source, + "char_count": len(text), + "chunk_count": chunk_index, + "findings": len(findings), + }) + return findings + + +def _walk_strings(value: Any, prefix: str) -> Iterable[tuple[str, str]]: + if isinstance(value, str): + yield prefix, value + elif isinstance(value, dict): + for key, child in value.items(): + yield from _walk_strings(child, f"{prefix}.{key}") + elif isinstance(value, list): + for index, child in enumerate(value): + yield from _walk_strings(child, f"{prefix}[{index}]") + + +def _session_id(session: dict[str, Any]) -> str | None: + for key in ("id", "session_id", "conversation_id"): + value = session.get(key) + if isinstance(value, str): + return value + return None + + +def scan_session( + session: dict[str, Any], + *, + device: str | None = None, + min_score: float = _DEFAULT_MIN_SCORE, + progress_callback: ProgressCallback | None = None, + include_tool_io: bool = True, + roles: set[str] | None = None, + model: str | None = None, +) -> list[Finding]: + found: list[Finding] = [] + session_id = _session_id(session) + source = session.get("source") if isinstance(session.get("source"), str) else None + + for msg_index, msg in enumerate(session.get("messages", [])): + if not isinstance(msg, dict): + continue + role = msg.get("role") + if roles is not None and (not isinstance(role, str) or role not in roles): + continue + # Mirror secrets.transform_session: walk content/thinking AND content_parts. + for field in ("content", "thinking", "content_parts"): + if field in msg: + prefix = f"messages[{msg_index}].{field}" + for path, text in _walk_strings(msg[field], prefix): + for finding in scan_text( + text, + device=device, + min_score=min_score, + progress_callback=progress_callback, + field=path, + session_id=session_id, + source=source, + model=model, + ): + found.append(_with_context(finding, path, session_id, source)) + if include_tool_io: + for tool_index, tool_use in enumerate(msg.get("tool_uses", [])): + if not isinstance(tool_use, dict): + continue + for field in ("input", "output"): + if field in tool_use: + prefix = f"messages[{msg_index}].tool_uses[{tool_index}].{field}" + for path, text in _walk_strings(tool_use[field], prefix): + for finding in scan_text( + text, + device=device, + min_score=min_score, + progress_callback=progress_callback, + field=path, + session_id=session_id, + source=source, + model=model, + ): + found.append(_with_context(finding, path, session_id, source)) + return found + + +def redact_text( + text: str, + *, + device: str | None = None, + min_score: float = _DEFAULT_MIN_SCORE, + progress_callback: ProgressCallback | None = None, + field: str | None = None, + session_id: str | None = None, + source: str | None = None, + model: str | None = None, +) -> tuple[str, list[Finding]]: + findings = scan_text( + text, + device=device, + min_score=min_score, + progress_callback=progress_callback, + field=field, + session_id=session_id, + source=source, + model=model, + ) + positioned = [f for f in findings if isinstance(f.start, int) and isinstance(f.end, int) and f.end > f.start] + if not positioned: + return text, findings + redacted = text + for finding in sorted(positioned, key=lambda f: int(f.start or 0), reverse=True): + start = int(finding.start or 0) + end = int(finding.end or start) + redacted = redacted[:start] + "[REDACTED]" + redacted[end:] + return redacted, findings + + +def _redact_strings_in_value( + value: Any, + prefix: str, + *, + device: str | None, + min_score: float, + progress_callback: ProgressCallback | None, + session_id: str | None, + source: str | None, + model: str | None = None, +) -> tuple[Any, list[Finding]]: + if isinstance(value, str): + redacted, findings = redact_text( + value, + device=device, + min_score=min_score, + progress_callback=progress_callback, + field=prefix, + session_id=session_id, + source=source, + model=model, + ) + return redacted, [_with_context(f, prefix, session_id, source) for f in findings] + if isinstance(value, dict): + out: dict[str, Any] = {} + findings: list[Finding] = [] + for key, child in value.items(): + redacted, child_findings = _redact_strings_in_value( + child, + f"{prefix}.{key}", + device=device, + min_score=min_score, + progress_callback=progress_callback, + session_id=session_id, + source=source, + model=model, + ) + out[key] = redacted + findings.extend(child_findings) + return out, findings + if isinstance(value, list): + out_list: list[Any] = [] + findings: list[Finding] = [] + for index, child in enumerate(value): + redacted, child_findings = _redact_strings_in_value( + child, + f"{prefix}[{index}]", + device=device, + min_score=min_score, + progress_callback=progress_callback, + session_id=session_id, + source=source, + model=model, + ) + out_list.append(redacted) + findings.extend(child_findings) + return out_list, findings + return value, [] + + +def _replace_strings(value: Any, replacement: str) -> Any: + if isinstance(value, str): + return replacement if value else value + if isinstance(value, dict): + return {key: _replace_strings(child, replacement) for key, child in value.items()} + if isinstance(value, list): + return [_replace_strings(child, replacement) for child in value] + return value + + +def _redact_oversized_session( + session: dict[str, Any], + *, + include_tool_io: bool, + roles: set[str] | None, + reason: str, +) -> tuple[dict[str, Any], list[Finding]]: + redacted_session = session + session_id = _session_id(redacted_session) + source = redacted_session.get("source") if isinstance(redacted_session.get("source"), str) else None + replacement = "[REDACTED: oversized session]" + redacted_fields = 0 + for msg in redacted_session.get("messages", []): + if not isinstance(msg, dict): + continue + role = msg.get("role") + if roles is not None and (not isinstance(role, str) or role not in roles): + continue + for field in ("content", "thinking", "content_parts"): + if field in msg: + msg[field] = _replace_strings(msg[field], replacement) + redacted_fields += 1 + if include_tool_io: + for tool_use in msg.get("tool_uses", []): + if not isinstance(tool_use, dict): + continue + for field in ("input", "output"): + if field in tool_use: + tool_use[field] = _replace_strings(tool_use[field], replacement) + redacted_fields += 1 + finding = Finding( + entity="OVERSIZED_SESSION_REDACTED", + text=reason, + score=1.0, + field=f"messages ({redacted_fields} fields)", + session_id=session_id, + source=source, + ) + return redacted_session, [finding] + + +def redact_session( + session: dict[str, Any], + *, + device: str | None = None, + min_score: float = _DEFAULT_MIN_SCORE, + progress_callback: ProgressCallback | None = None, + include_tool_io: bool = True, + roles: set[str] | None = None, + model: str | None = None, +) -> tuple[dict[str, Any], list[Finding]]: + redacted_session = json.loads(json.dumps(session)) + + # Oversized-session guard: blanket-redact sessions too large to run the model + # over safely (avoids pathological memory/time blowups). This protects the + # in-memory export path the same way the old shard pipeline did. + string_count, char_count = _string_stats(redacted_session.get("messages", [])) + if char_count > _MAX_MODEL_SESSION_CHARS or string_count > _MAX_MODEL_SESSION_STRINGS: + reason = ( + f"session exceeds model privacy-filter guard " + f"({char_count} chars, {string_count} strings)" + ) + return _redact_oversized_session( + redacted_session, + include_tool_io=include_tool_io, + roles=roles, + reason=reason, + ) + + found: list[Finding] = [] + session_id = _session_id(redacted_session) + source = redacted_session.get("source") if isinstance(redacted_session.get("source"), str) else None + + for msg_index, msg in enumerate(redacted_session.get("messages", [])): + if not isinstance(msg, dict): + continue + role = msg.get("role") + if roles is not None and (not isinstance(role, str) or role not in roles): + continue + # Mirror secrets.transform_session: walk content/thinking AND content_parts. + for field in ("content", "thinking", "content_parts"): + if field in msg: + prefix = f"messages[{msg_index}].{field}" + msg[field], findings = _redact_strings_in_value( + msg[field], + prefix, + device=device, + min_score=min_score, + progress_callback=progress_callback, + session_id=session_id, + source=source, + model=model, + ) + found.extend(findings) + if include_tool_io: + for tool_index, tool_use in enumerate(msg.get("tool_uses", [])): + if not isinstance(tool_use, dict): + continue + for field in ("input", "output"): + if field in tool_use: + prefix = f"messages[{msg_index}].tool_uses[{tool_index}].{field}" + tool_use[field], findings = _redact_strings_in_value( + tool_use[field], + prefix, + device=device, + min_score=min_score, + progress_callback=progress_callback, + session_id=session_id, + source=source, + model=model, + ) + found.extend(findings) + return redacted_session, found + + +def _with_context(finding: Finding, field: str, session_id: str | None, source: str | None) -> Finding: + return replace(finding, field=field, session_id=session_id, source=source) + + +def _string_stats(value: Any) -> tuple[int, int]: + string_count = 0 + char_count = 0 + for _, text in _walk_strings(value, ""): + string_count += 1 + char_count += len(text) + return string_count, char_count + + +def diff_findings( + findings: Iterable[Finding], + known_findings: dict[str, Any] | None, +) -> tuple[list[Finding], list[Finding]]: + known = set((known_findings or {}).keys()) + new: list[Finding] = [] + old: list[Finding] = [] + for finding in findings: + if finding.fingerprint() in known: + old.append(finding) + else: + new.append(finding) + return new, old + + +def record_findings( + findings: Iterable[Finding], + known_findings: dict[str, Any] | None = None, +) -> dict[str, Any]: + from datetime import datetime, timezone + + registry = dict(known_findings or {}) + now = datetime.now(tz=timezone.utc).isoformat() + for finding in findings: + fp = finding.fingerprint() + existing = registry.get(fp) if isinstance(registry.get(fp), dict) else {} + first_seen = existing.get("first_seen") or now + registry[fp] = { + "entity": finding.entity, + "text": finding.text, + "first_seen": first_seen, + "last_seen": now, + "count": int(existing.get("count", 0)) + 1, + } + return registry diff --git a/docs/reliability-merge-and-privacy-filter-plan.md b/docs/reliability-merge-and-privacy-filter-plan.md index 6b4a1c7..e560109 100644 --- a/docs/reliability-merge-and-privacy-filter-plan.md +++ b/docs/reliability-merge-and-privacy-filter-plan.md @@ -111,12 +111,17 @@ Bugs / gaps (line refs on the `5d0a741` version): - Minor: cross-chunk-boundary entities may be split/under-redacted (low severity); `min_score` discarded in `_load` then re-applied in `scan_text` (cosmetic); `dtype=` kwarg needs transformers ≥4.45. -### Model recommendation (drop-in for the existing pipeline call) -- **Default: `iiiorg/piiranha-v1-detect-personal-information`** — MIT, ~280 MB, DeBERTa - token-classification, MPS-ok, drops in with **zero adapter code**. -- **Fallback: `lakshyakh93/deberta_finetuned_pii`** — broader labels, also a pure drop-in. -- Avoid GLiNER / Presidio for v1 (need adapters / heavier deps). Verify Hub availability and that - `aggregation_strategy="simple"` yields `start`/`end` offsets before committing. +### Model decision (drop-in for the existing pipeline call) +- **Default: `openai/privacy-filter`** — the model the original implementation + intended. It is a REAL, published model (Apache-2.0, ~300k downloads, + token-classification, safetensors + ONNX) — the earlier "placeholder that doesn't + exist" claim was WRONG. It drops into the existing + `pipeline(task="token-classification", aggregation_strategy="simple")` call with no + adapter code and detects PII as intended. +- Overridable via the `privacy_filter.model` config key / `DATACLAW_PRIVACY_FILTER_MODEL` + env var, so swapping is trivial. +- Alternatives if ever needed: `lakshyakh93/deberta_finetuned_pii` (MIT, broad labels, + pure drop-in). Avoid GLiNER / Presidio (need adapters / heavier deps). ### Rewiring against HEAD `_cli` structure - **Mutation happens only at EXPORT time**, inline per session, right after diff --git a/pyproject.toml b/pyproject.toml index 968e9a9..13a37dc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,6 +38,14 @@ dataclaw = "dataclaw.cli:main" [project.optional-dependencies] dev = ["pre-commit", "pytest", "ruff"] build = ["pyinstaller>=6.10", "keyring>=25", "secretstorage>=3.3; sys_platform == 'linux'"] +# Optional model-based PII filter (power-user / first-run model download). +# Deliberately NOT a core dep and NOT bundled in the Mac sidecar by default. +pii = ["transformers>=4.57", "torch>=2.3", "accelerate", "tokenizers"] + +[tool.pytest.ini_options] +markers = [ + "pii: tests that require the optional model PII stack (transformers/torch); skipped when absent", +] [tool.ruff] line-length = 120 diff --git a/tests/conftest.py b/tests/conftest.py index c8007fe..d0c404b 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -66,3 +66,16 @@ def tmp_config(tmp_path, monkeypatch): monkeypatch.setattr("dataclaw.config.CONFIG_DIR", config_dir) monkeypatch.setattr("dataclaw.config.CONFIG_FILE", config_file) return config_file + + +@pytest.fixture(autouse=True) +def _isolate_user_config(tmp_path_factory, monkeypatch): + """Never read the developer's real ~/.dataclaw/config.json during tests. + + Without this, settings like ``privacy_filter.enabled`` from a real local + config would leak into export tests (slow, non-deterministic). Tests that + need specific config still monkeypatch ``load_config`` or use ``tmp_config``. + """ + isolated_dir = tmp_path_factory.mktemp("dataclaw-config") + monkeypatch.setattr("dataclaw.config.CONFIG_DIR", isolated_dir) + monkeypatch.setattr("dataclaw.config.CONFIG_FILE", isolated_dir / "config.json") diff --git a/tests/test_privacy_filter.py b/tests/test_privacy_filter.py new file mode 100644 index 0000000..5afa073 --- /dev/null +++ b/tests/test_privacy_filter.py @@ -0,0 +1,382 @@ +"""Tests for optional privacy-filter scanning.""" + +import builtins + +import pytest + +from dataclaw import privacy_filter as pf + + +def test_fingerprint_deterministic(): + first = pf.Finding("NAME", "Jane Doe", 0.91, start=1, end=9) + second = pf.Finding("NAME", "Jane Doe", 0.12, start=40, end=48) + + assert first.fingerprint() == second.fingerprint() + + +def test_diff_findings_splits_correctly(): + findings = [ + pf.Finding("NAME", "A", 0.9), + pf.Finding("NAME", "B", 0.9), + pf.Finding("ORG", "C", 0.9), + pf.Finding("ORG", "D", 0.9), + ] + known = { + findings[1].fingerprint(): {"count": 1}, + findings[3].fingerprint(): {"count": 1}, + } + + new, old = pf.diff_findings(findings, known) + + assert [f.text for f in new] == ["A", "C"] + assert [f.text for f in old] == ["B", "D"] + + +def test_record_findings_increments(): + finding = pf.Finding("NAME", "Jane Doe", 0.9) + + registry = pf.record_findings([finding]) + first_seen = registry[finding.fingerprint()]["first_seen"] + registry = pf.record_findings([finding], registry) + + record = registry[finding.fingerprint()] + assert record["count"] == 2 + assert record["first_seen"] == first_seen + assert record["last_seen"] >= first_seen + + +def test_is_available_false_without_deps(monkeypatch): + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "transformers": + raise ImportError("No module named transformers") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", fake_import) + + assert pf.is_available() is False + + +def test_scan_text_passes_device_and_min_score_through(monkeypatch): + seen = {} + + def fake_load(**kwargs): + seen.update(kwargs) + return lambda _text: [ + {"entity_group": "NAME", "word": "Jane", "score": 0.8, "start": 0, "end": 4}, + ] + + monkeypatch.setattr(pf, "_load", fake_load) + + findings = pf.scan_text("Jane", device="mps", min_score=0.7) + + assert seen == {"device": "mps", "min_score": 0.7, "model": None} + assert findings[0].text == "Jane" + + +def test_scan_session_aggregates_messages_and_thinking(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("ORG", "ACME Corp")) + session = { + "messages": [ + {"content": "hello"}, + {"thinking": "Need to call ACME Corp before shipping."}, + ], + } + + findings = pf.scan_session(session) + + assert len(findings) == 1 + assert findings[0].field.startswith("messages[") + assert findings[0].field.endswith(".thinking") + + +def test_scan_session_walks_nested_tool_output_shapes(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = { + "messages": [{ + "tool_uses": [ + {"output": {"files": [{"content": "Jane Doe"}], "stdout": "Jane Doe"}}, + {"output": "Jane Doe"}, + {"input": {"command": "echo Jane Doe"}}, + ], + }], + } + + fields = {finding.field for finding in pf.scan_session(session)} + + assert "messages[0].tool_uses[0].output.files[0].content" in fields + assert "messages[0].tool_uses[1].output" in fields + assert "messages[0].tool_uses[2].input.command" in fields + + +def test_scan_session_can_skip_tool_io(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = { + "messages": [{ + "content": "No match here", + "tool_uses": [{"output": "Jane Doe"}], + }], + } + + assert pf.scan_session(session, include_tool_io=False) == [] + + +def test_redact_session_replaces_model_findings(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = {"messages": [{"content": "Ask Jane Doe about this."}]} + + redacted, findings = pf.redact_session(session) + + assert len(findings) == 1 + assert redacted["messages"][0]["content"] == "Ask [REDACTED] about this." + assert session["messages"][0]["content"] == "Ask Jane Doe about this." + + +def test_redact_session_can_limit_roles(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = { + "messages": [ + {"role": "assistant", "content": "Jane Doe from assistant"}, + {"role": "user", "content": "Jane Doe from user"}, + ], + } + + redacted, findings = pf.redact_session(session, roles={"user"}) + + assert len(findings) == 1 + assert redacted["messages"][0]["content"] == "Jane Doe from assistant" + assert redacted["messages"][1]["content"] == "[REDACTED] from user" + + +def test_redact_session_redacts_oversized_sessions_without_model(monkeypatch): + monkeypatch.setattr(pf, "_MAX_MODEL_SESSION_CHARS", 10) + monkeypatch.setattr(pf, "_MAX_MODEL_SESSION_STRINGS", 10) + load_calls = [] + monkeypatch.setattr(pf, "_load", lambda **kwargs: load_calls.append(kwargs) or (lambda _text: [])) + session = { + "session_id": "s1", + "source": "codex", + "messages": [{"role": "user", "content": "very sensitive long text"}], + } + + redacted, findings = pf.redact_session(session) + + # The model is never loaded for an oversized session; it is blanket-redacted. + assert redacted["messages"][0]["content"] == "[REDACTED: oversized session]" + assert findings[0].entity == "OVERSIZED_SESSION_REDACTED" + assert load_calls == [] + # Original session dict is untouched (redact_session deep-copies). + assert session["messages"][0]["content"] == "very sensitive long text" + + +def test_redact_session_redacts_content_parts(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = { + "messages": [{ + "content": "no match", + "content_parts": [ + {"type": "text", "text": "Contact Jane Doe today"}, + {"type": "image", "url": "https://example.com/x.png"}, + ], + }], + } + + redacted, findings = pf.redact_session(session) + + # PII inside content_parts (which secrets.transform_session also walks) is redacted. + assert redacted["messages"][0]["content_parts"][0]["text"] == "Contact [REDACTED] today" + assert any(f.field and ".content_parts" in f.field for f in findings) + + +def test_scan_session_walks_content_parts(monkeypatch): + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + session = { + "messages": [{ + "content_parts": [{"type": "text", "text": "Ask Jane Doe"}], + }], + } + + fields = {finding.field for finding in pf.scan_session(session)} + + assert "messages[0].content_parts[0].text" in fields + + +def test_dtype_auto_selects_bfloat16_on_mps(monkeypatch): + torch = pytest.importorskip("torch") + + # Force the auto path even if a user has overridden dtype in their config. + monkeypatch.setattr(pf, "_config_dtype", lambda: "auto") + # Pretend we're on Apple Silicon with MPS available. + monkeypatch.setattr(torch.backends.mps, "is_available", lambda: True) + # Reset the pipeline cache so the next _load triggers a fresh build. + monkeypatch.setattr(pf, "_PIPELINES", {}) + + captured: dict[str, object] = {} + + def fake_build(**kwargs): + captured.update(kwargs) + return lambda _text: [] + + monkeypatch.setattr(pf, "_build_pipeline", fake_build) + + pf._load(device="mps") + + assert captured.get("dtype") is torch.bfloat16 + assert captured.get("model") == pf.MODEL_ID + assert captured.get("device") == "mps" + + +def test_load_defaults_to_mps_when_available(monkeypatch): + torch = pytest.importorskip("torch") + + monkeypatch.setattr(pf, "_config_device", lambda: "auto") + monkeypatch.setattr(pf, "_config_dtype", lambda: "auto") + monkeypatch.setattr(torch.backends.mps, "is_available", lambda: True) + monkeypatch.setattr(pf, "_PIPELINES", {}) + + captured: dict[str, object] = {} + + def fake_build(**kwargs): + captured.update(kwargs) + return lambda _text: [] + + monkeypatch.setattr(pf, "_build_pipeline", fake_build) + + pf._load() + + assert captured.get("device") == "mps" + assert captured.get("dtype") is torch.bfloat16 + + +def test_dtype_auto_falls_back_to_fp32_on_cpu(monkeypatch): + torch = pytest.importorskip("torch") + + monkeypatch.setattr(pf, "_config_dtype", lambda: "auto") + monkeypatch.setattr(torch.backends.mps, "is_available", lambda: False) + monkeypatch.setattr(pf, "_PIPELINES", {}) + + captured: dict[str, object] = {} + + def fake_build(**kwargs): + captured.update(kwargs) + return lambda _text: [] + + monkeypatch.setattr(pf, "_build_pipeline", fake_build) + + pf._load() + + assert captured.get("dtype") is torch.float32 + + +@pytest.mark.pii +def test_pipeline_runs_end_to_end_against_real_model(): + # Live smoke test: proves the default model id loads and the + # token-classification pipeline (aggregation_strategy="simple") runs + # end-to-end and returns offset-bearing findings. Detection *quality* of a + # specific checkpoint under a specific transformers version is an open + # verification item (see the Part B design doc), so we only assert the + # integration contract here, not a minimum hit count. + pytest.importorskip("transformers") + pytest.importorskip("torch") + + findings = pf.scan_text("Hi, my name is John Smith and my email is john@example.com") + + assert isinstance(findings, list) + for finding in findings: + assert finding.start is None or isinstance(finding.start, int) + assert finding.end is None or isinstance(finding.end, int) + + +def test_config_model_env_override(monkeypatch): + monkeypatch.setenv("DATACLAW_PRIVACY_FILTER_MODEL", "acme/custom-pii") + assert pf._config_model() == "acme/custom-pii" + + +def test_config_model_defaults_to_model_id(monkeypatch): + monkeypatch.delenv("DATACLAW_PRIVACY_FILTER_MODEL", raising=False) + monkeypatch.setattr(pf, "_config_model", pf._config_model) # no-op, keep real fn + # With no config file / no override, falls back to the default model id. + monkeypatch.setattr("dataclaw.config.load_config", lambda: {}) + assert pf._config_model() == pf.MODEL_ID == pf.DEFAULT_MODEL_ID + + +# --- export-time wiring (graceful degradation) --------------------------------- + +from dataclaw._cli import exporting as _exp # noqa: E402 + + +def test_apply_model_privacy_filter_disabled_is_noop(): + session = {"messages": [{"content": "Jane Doe"}]} + cfg = _exp._PrivacyFilterConfig(enabled=False) + + assert _exp._apply_model_privacy_filter(session, cfg) is session + + +def test_apply_model_privacy_filter_graceful_when_unavailable(monkeypatch, capsys): + # Force a fresh warning each run. + monkeypatch.setattr(_exp, "_PF_WARNED", False) + # Simulate torch/transformers absent. + monkeypatch.setattr("dataclaw.privacy_filter.is_available", lambda: False) + + session = {"messages": [{"role": "user", "content": "Mechanical [REDACTED] intact"}]} + cfg = _exp._PrivacyFilterConfig(enabled=True) + + result = _exp._apply_model_privacy_filter(session, cfg) + + # Export does not crash; mechanical redaction is preserved unchanged. + assert result == session + assert result["messages"][0]["content"] == "Mechanical [REDACTED] intact" + err = capsys.readouterr().err + assert "model privacy filter skipped" in err + + +def test_apply_model_privacy_filter_runs_when_available(monkeypatch): + monkeypatch.setattr("dataclaw.privacy_filter.is_available", lambda: True) + monkeypatch.setattr(pf, "_load", lambda **_kw: _match_pipe("NAME", "Jane Doe")) + + session = {"messages": [{"role": "user", "content": "Call Jane Doe now"}]} + cfg = _exp._PrivacyFilterConfig(enabled=True, min_score=0.5) + + result = _exp._apply_model_privacy_filter(session, cfg) + + assert result["messages"][0]["content"] == "Call [REDACTED] now" + + +def test_read_privacy_filter_config(monkeypatch): + monkeypatch.setattr( + "dataclaw.config.load_config", + lambda: {"privacy_filter": {"enabled": True, "device": "mps", "min_score": 0.7, "model": "acme/m"}}, + ) + cfg = _exp._read_privacy_filter_config() + + assert cfg.enabled is True + assert cfg.device == "mps" + assert cfg.min_score == 0.7 + assert cfg.model == "acme/m" + + +def test_read_privacy_filter_config_defaults(monkeypatch): + monkeypatch.setattr("dataclaw.config.load_config", lambda: {}) + cfg = _exp._read_privacy_filter_config() + + assert cfg.enabled is False + assert cfg.device is None + assert cfg.min_score == 0.85 + assert cfg.model is None + + +def _match_pipe(entity: str, needle: str): + def pipe(text): + start = text.find(needle) + if start < 0: + return [] + return [{ + "entity_group": entity, + "word": needle, + "score": 0.99, + "start": start, + "end": start + len(needle), + }] + + return pipe From c2b703d064a5a14e07eecac3e69b74e16dff9b24 Mon Sep 17 00:00:00 2001 From: POM Date: Sat, 30 May 2026 02:12:10 +0200 Subject: [PATCH 3/5] Close two gaps found in adversarial sense-check of the new code 1. Carried-forward remote records now get the model privacy filter, not just mechanical redaction. _build_carry_forward_redactor reads the privacy-filter config and applies _apply_model_privacy_filter after transform_session, mirroring the export loop. Without this, the bulk of a steady-state push (carry-forwards) shipped with mechanical redaction only when the model filter was enabled. 2. merge_jsonl_union counts remote_total/local_total by UNIQUE merge key instead of raw line count. A remote file from the old non-deduping uploader could contain duplicate (source, session_id) lines; counting raw lines made merged_total < remote_total trip the union-invariant guard and permanently block publishing even though no session was lost. Tests: duplicate-remote-key invariant; carry-forward applies model filter. Co-Authored-By: Claude Opus 4.8 (1M context) --- dataclaw/_cli/exporting.py | 7 +++++++ dataclaw/jsonl_tools.py | 9 ++++++++- tests/test_jsonl_merge.py | 23 +++++++++++++++++++++++ tests/test_privacy_filter.py | 22 ++++++++++++++++++++++ 4 files changed, 60 insertions(+), 1 deletion(-) diff --git a/dataclaw/_cli/exporting.py b/dataclaw/_cli/exporting.py index 383caca..4f3b731 100644 --- a/dataclaw/_cli/exporting.py +++ b/dataclaw/_cli/exporting.py @@ -760,6 +760,10 @@ def _build_carry_forward_redactor(redaction: dict | None): redaction = redaction or {} extra_usernames = list(redaction.get("redact_usernames") or []) custom_strings = list(redaction.get("redact_strings") or []) + # Carried-forward records must get the SAME passes fresh local sessions get, + # including the model privacy filter when it is enabled — otherwise the bulk of + # a steady-state push (carry-forwards) would ship with mechanical redaction only. + pf_config = _read_privacy_filter_config() def redact_fn(record: dict) -> dict: # A fresh Anonymizer per record matches the per-worker construction and keeps @@ -772,6 +776,9 @@ def redact_fn(record: dict) -> dict: custom_strings=custom_strings, non_anon_string_keys=get_provider_non_anon_string_keys(source), ) + # Mirror the export loop: model PII pass after mechanical redaction. No-op + # (and idempotent) when the filter is disabled or unavailable. + redacted = _apply_model_privacy_filter(redacted, pf_config) return redacted return redact_fn diff --git a/dataclaw/jsonl_tools.py b/dataclaw/jsonl_tools.py index af78434..2e7ee96 100644 --- a/dataclaw/jsonl_tools.py +++ b/dataclaw/jsonl_tools.py @@ -239,7 +239,12 @@ def merge_jsonl_union( remote_records = _load_raw_records(remote_path) local_records = _load_raw_records(local_path) - stats = MergeStats(remote_total=len(remote_records), local_total=len(local_records)) + # Totals are counted by UNIQUE merge key, not raw line count. A remote file + # written by the old non-deduping uploader can contain duplicate + # (source, session_id) lines; counting raw lines would make merged_total < + # remote_total trip the union-invariant guard and permanently block publishing + # even though no session was dropped. (Set precisely after the build below.) + stats = MergeStats() # Build winning record per key, tracking origin (which side won) and which sides # the key appeared on, so we can classify the change and re-redact correctly. @@ -296,6 +301,8 @@ def merge_jsonl_union( handle.write(b"\n") merged_total += 1 + stats.remote_total = len(in_remote) + stats.local_total = len(in_local) stats.merged_total = merged_total return stats diff --git a/tests/test_jsonl_merge.py b/tests/test_jsonl_merge.py index e685230..e4b981c 100644 --- a/tests/test_jsonl_merge.py +++ b/tests/test_jsonl_merge.py @@ -208,6 +208,29 @@ def test_union_invariant_merged_ge_remote(self, tmp_path): assert merged == {"r1", "r2"} assert stats.merged_total >= stats.remote_total + def test_duplicate_remote_keys_do_not_trip_invariant(self, tmp_path): + # The old non-deduping uploader could leave duplicate (source, session_id) + # lines in the remote. remote_total must count UNIQUE keys, not raw lines, + # so dedup never makes merged_total < remote_total (which would permanently + # block publishing). + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + _write_jsonl( + remote, + [ + {"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}, + {"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}, # dup line + ], + ) + _write_jsonl(local, []) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + assert stats.remote_total == 1 # unique keys, not 2 raw lines + assert stats.merged_total == 1 + assert stats.merged_total >= stats.remote_total # invariant holds, no false abort + def test_changelog_line_format(self): stats = MergeStats( remote_total=3, local_total=2, merged_total=4, added=1, updated=1, carried_forward=2, unchanged=0 diff --git a/tests/test_privacy_filter.py b/tests/test_privacy_filter.py index 5afa073..d68cf4c 100644 --- a/tests/test_privacy_filter.py +++ b/tests/test_privacy_filter.py @@ -366,6 +366,28 @@ def test_read_privacy_filter_config_defaults(monkeypatch): assert cfg.model is None +def test_carry_forward_redactor_applies_model_filter(monkeypatch): + # Carried-forward remote records must get the model PII pass too, not just + # mechanical redaction -- otherwise the bulk of a steady-state push ships + # without model scrubbing. + monkeypatch.setattr(_exp, "_read_privacy_filter_config", lambda: _exp._PrivacyFilterConfig(enabled=True)) + + seen = [] + + def fake_model_filter(session, pf_config): + seen.append(pf_config.enabled) + session["_model_filtered"] = True + return session + + monkeypatch.setattr(_exp, "_apply_model_privacy_filter", fake_model_filter) + + redact_fn = _exp._build_carry_forward_redactor({"redact_strings": [], "redact_usernames": []}) + out = redact_fn({"source": "claude", "session_id": "r1", "messages": [{"role": "user", "content": "hi"}]}) + + assert out.get("_model_filtered") is True + assert seen == [True] + + def _match_pipe(entity: str, needle: str): def pipe(text): start = text.find(needle) From 619ef63a848d2afc33d6b499ab387e21bd2452a5 Mon Sep 17 00:00:00 2001 From: POM Date: Sat, 30 May 2026 02:24:44 +0200 Subject: [PATCH 4/5] Make dataset updates O(new data): policy stamp + skip-unchanged + corrupt-line guard Addresses the scale/elegance findings from the merge review. Keystone - redaction-policy stamp: - Stamp each exported record with redaction_policy_version() (hash of redact strings/usernames + model config + a code version). Computed in both export paths (serial + parallel worker), written after fingerprinting. - The merge's carry-forward redactor now SKIPS re-redaction for any record already stamped with the current version, re-redacting (and re-stamping) only stale ones. This turns a steady-state push from O(total history) -- which re-ran the PII model over the whole dataset every push -- into ~O(new data), while preserving the tighten-only guarantee (policy change bumps the version and forces a one-time full re-scan). Cheap wins: - Skip ALL uploads when the merged file is byte-identical to the remote we just downloaded, so a no-change push stops churning the repo with empty commits (the metadata timestamp alone otherwise differed every run). - _load_raw_records preserves unparseable JSONL lines verbatim instead of raising: a corrupt remote line can no longer drop data or permanently wedge all future pushes. Counted in MergeStats.malformed_preserved. Tests: policy-version sensitivity, carry-forward stamp-skip vs re-redact, malformed-line preservation + invariant. Co-Authored-By: Claude Opus 4.8 (1M context) --- dataclaw/_cli/exporting.py | 66 ++++++++++++++++++++++++++++++++++++ dataclaw/jsonl_tools.py | 52 ++++++++++++++++++++++------ tests/test_jsonl_merge.py | 20 +++++++++++ tests/test_privacy_filter.py | 38 +++++++++++++++++++++ 4 files changed, 165 insertions(+), 11 deletions(-) diff --git a/dataclaw/_cli/exporting.py b/dataclaw/_cli/exporting.py index 4f3b731..cb98749 100644 --- a/dataclaw/_cli/exporting.py +++ b/dataclaw/_cli/exporting.py @@ -199,6 +199,33 @@ def _read_privacy_filter_config() -> _PrivacyFilterConfig: return _PrivacyFilterConfig(enabled=enabled, device=device, min_score=min_score, model=model) +# Bump when the redaction CODE changes in a way that should force re-scanning +# already-published records (independent of user config). +_REDACTION_CODE_VERSION = "1" +_REDACTION_POLICY_KEY = "redaction_policy" + + +def redaction_policy_version(custom_strings, extra_usernames, pf_config: _PrivacyFilterConfig) -> str: + """Stable fingerprint of the CURRENT redaction policy. + + Records are stamped with this on export. The merge re-redacts a + carried-forward record only when its stamp differs from the current version, + so a steady-state push is ~O(new data) instead of re-running redaction (and + the PII model) over the entire history every push. Tightening the policy + (new redact strings/usernames, model change, code bump) changes the version, + which forces a one-time full re-scan -- preserving the tighten-only guarantee. + """ + parts = [ + _REDACTION_CODE_VERSION, + "\x1f".join(sorted(custom_strings or [])), + "\x1f".join(sorted(extra_usernames or [])), + "1" if pf_config.enabled else "0", + (pf_config.model or "") if pf_config.enabled else "", + format(pf_config.min_score, ".4f") if pf_config.enabled else "", + ] + return hashlib.sha256("\x1e".join(parts).encode("utf-8")).hexdigest()[:16] + + _PF_WARNED = False @@ -290,6 +317,9 @@ def _export_session_task_worker(payload) -> _WorkerSessionResult: # Optional model privacy filter runs AFTER mechanical redaction, before hashing. session = _apply_model_privacy_filter(session, pf_config) fingerprint = _gemini_dedupe_fingerprint(session, task.source) + # Stamp the current redaction policy so future merges can skip re-redacting + # this record while it stays current. After fingerprinting so dedup is stable. + session[_REDACTION_POLICY_KEY] = redaction_policy_version(custom_strings, extra_usernames, pf_config) stats = session.get("stats", {}) input_tokens, output_tokens = _token_totals(stats) has_token_stats = isinstance(stats, dict) and ("input_tokens" in stats or "output_tokens" in stats) @@ -354,6 +384,7 @@ def _export_to_jsonl_serial( total_input_tokens = 0 total_output_tokens = 0 seen_fingerprints: set[str] = set() + policy_version = redaction_policy_version(custom_strings, _export_extra_usernames(anonymizer), pf_config) for project in selected_projects: print(f" Parsing {project['display_name']}...", end="", flush=True) @@ -392,6 +423,7 @@ def _export_to_jsonl_serial( if fingerprint is not None: seen_fingerprints.add(fingerprint) + session[_REDACTION_POLICY_KEY] = policy_version fh.write(json.dumps_bytes(session)) fh.write(b"\n") total += 1 @@ -764,8 +796,16 @@ def _build_carry_forward_redactor(redaction: dict | None): # including the model privacy filter when it is enabled — otherwise the bulk of # a steady-state push (carry-forwards) would ship with mechanical redaction only. pf_config = _read_privacy_filter_config() + policy_version = redaction_policy_version(custom_strings, extra_usernames, pf_config) def redact_fn(record: dict) -> dict: + # Keystone optimization: a record already redacted under the CURRENT policy + # needs no re-work. Skip it (return verbatim) so a steady-state push doesn't + # re-run mechanical + model redaction over the entire history every time. + # The version changes whenever the policy tightens, forcing a one-time + # re-scan -- so the tighten-only guarantee is preserved. + if record.get(_REDACTION_POLICY_KEY) == policy_version: + return record # A fresh Anonymizer per record matches the per-worker construction and keeps # pseudonyms deterministic (hash-based), so re-redaction stays idempotent. anonymizer = Anonymizer(extra_usernames=extra_usernames) @@ -779,6 +819,7 @@ def redact_fn(record: dict) -> dict: # Mirror the export loop: model PII pass after mechanical redaction. No-op # (and idempotent) when the filter is disabled or unavailable. redacted = _apply_model_privacy_filter(redacted, pf_config) + redacted[_REDACTION_POLICY_KEY] = policy_version return redacted return redact_fn @@ -855,6 +896,16 @@ def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict, redaction: d meta["sessions"] = merged_total upload_target = merged_path if remote_path is not None else jsonl_path + # No-op detection: if the merged result is byte-identical to the remote + # we just downloaded, there is nothing new to publish. Skip ALL uploads + # (data + metadata + README) so a no-change push doesn't churn the repo + # with empty commits (the metadata timestamp alone would otherwise differ + # every run). Only reachable when remote exists (passthrough has no remote). + if remote_path is not None and _files_identical(merged_path, remote_path): + print("Dataset already up to date; nothing to publish.") + print(f"\nDataset: {hf_dataset_url(repo_id)}") + return + try: api.upload_file( path_or_fileobj=str(upload_target), @@ -897,6 +948,21 @@ def push_to_huggingface(jsonl_path: Path, repo_id: str, meta: dict, redaction: d print(f"Browse all: {hf_browse_tagged_url()}") +def _files_identical(a: Path, b: Path) -> bool: + """True if two files have identical bytes (chunked sha256, large-file safe).""" + if a.stat().st_size != b.stat().st_size: + return False + + def _digest(path: Path) -> str: + hasher = hashlib.sha256() + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + hasher.update(chunk) + return hasher.hexdigest() + + return _digest(a) == _digest(b) + + def _merge_or_passthrough(remote_path: Path | None, local_path: Path, merged_path: Path, redact_fn) -> int: """Merge remote+local (or pass through local when remote is empty); return total.""" if remote_path is None: diff --git a/dataclaw/jsonl_tools.py b/dataclaw/jsonl_tools.py index 2e7ee96..75c4903 100644 --- a/dataclaw/jsonl_tools.py +++ b/dataclaw/jsonl_tools.py @@ -146,13 +146,17 @@ class MergeStats: updated: int = 0 # records present in both where the local copy won (superset) carried_forward: int = 0 # remote-only records preserved (re-redacted) unchanged: int = 0 # records present in both where the remote copy won (re-redacted) + malformed_preserved: int = 0 # unparseable lines carried through verbatim def changelog_line(self) -> str: - return ( + line = ( f"Merge: added {self.added}, updated {self.updated}, " f"carried_forward {self.carried_forward}, unchanged {self.unchanged} " f"(remote {self.remote_total} -> merged {self.merged_total})" ) + if self.malformed_preserved: + line += f"; preserved {self.malformed_preserved} unparseable line(s) verbatim" + return line def merge_identity_key(obj: dict[str, Any]) -> tuple[Any, ...]: @@ -199,18 +203,31 @@ def _record_prefers(candidate: dict[str, Any], current: dict[str, Any]) -> bool: return _end_time(candidate) > _end_time(current) -def _load_raw_records(path: Path) -> list[dict[str, Any]]: - """Load JSONL records RAW (no diff normalization, preserves ``originalFile``).""" +def _load_raw_records(path: Path) -> tuple[list[dict[str, Any]], list[bytes]]: + """Load JSONL records RAW (no diff normalization, preserves ``originalFile``). + + Returns ``(records, malformed_lines)``. A line that is not valid JSON (or not a + JSON object) is returned verbatim in ``malformed_lines`` rather than raising or + being dropped: the merge preserves it so a single corrupt remote line can never + silently lose data nor permanently wedge all future pushes. + """ records: list[dict[str, Any]] = [] + malformed: list[bytes] = [] with path.open("rb") as handle: for line in handle: - line = line.strip() - if not line: + stripped = line.strip() + if not stripped: + continue + try: + obj = orjson.loads(stripped) + except ValueError: # orjson.JSONDecodeError subclasses ValueError + malformed.append(stripped) continue - obj = orjson.loads(line) if isinstance(obj, dict): records.append(obj) - return records + else: + malformed.append(stripped) + return records, malformed def merge_jsonl_union( @@ -236,8 +253,11 @@ def merge_jsonl_union( ``redact_fn`` is injected so this module stays free of import cycles with the redaction pipeline. """ - remote_records = _load_raw_records(remote_path) - local_records = _load_raw_records(local_path) + remote_records, remote_malformed = _load_raw_records(remote_path) + local_records, local_malformed = _load_raw_records(local_path) + # Preserve unparseable lines verbatim (deduped by exact bytes) so a corrupt + # remote line is never dropped (data loss) nor allowed to abort the push (wedge). + malformed = list(dict.fromkeys(remote_malformed + local_malformed)) # Totals are counted by UNIQUE merge key, not raw line count. A remote file # written by the old non-deduping uploader can contain duplicate @@ -301,9 +321,19 @@ def merge_jsonl_union( handle.write(b"\n") merged_total += 1 - stats.remote_total = len(in_remote) - stats.local_total = len(in_local) + for raw in malformed: + handle.write(raw) + handle.write(b"\n") + merged_total += 1 + + # Malformed remote lines count toward remote_total (by UNIQUE bytes, matching + # the unique-key counting above) so the union invariant merged_total >= + # remote_total accounts for the records they represent without false-tripping + # on duplicate corrupt lines. + stats.remote_total = len(in_remote) + len(set(remote_malformed)) + stats.local_total = len(in_local) + len(set(local_malformed)) stats.merged_total = merged_total + stats.malformed_preserved = len(malformed) return stats diff --git a/tests/test_jsonl_merge.py b/tests/test_jsonl_merge.py index e4b981c..77b51cb 100644 --- a/tests/test_jsonl_merge.py +++ b/tests/test_jsonl_merge.py @@ -231,6 +231,26 @@ def test_duplicate_remote_keys_do_not_trip_invariant(self, tmp_path): assert stats.merged_total == 1 assert stats.merged_total >= stats.remote_total # invariant holds, no false abort + def test_malformed_remote_line_preserved_not_dropped(self, tmp_path): + # A corrupt remote line must be carried through verbatim (never dropped = + # data loss, never raised = wedged pushes). + remote = tmp_path / "remote.jsonl" + local = tmp_path / "local.jsonl" + out = tmp_path / "merged.jsonl" + remote.write_bytes( + orjson.dumps({"source": "claude", "session_id": "r1", "messages": [{"role": "user"}]}) + + b"\n" + + b"{this is not valid json,,,}\n" + ) + _write_jsonl(local, []) + + stats = merge_jsonl_union(remote, local, out, redact_fn=_identity) + + raw = out.read_bytes() + assert b"{this is not valid json,,,}" in raw # preserved verbatim + assert stats.malformed_preserved == 1 + assert stats.merged_total >= stats.remote_total # invariant still holds + def test_changelog_line_format(self): stats = MergeStats( remote_total=3, local_total=2, merged_total=4, added=1, updated=1, carried_forward=2, unchanged=0 diff --git a/tests/test_privacy_filter.py b/tests/test_privacy_filter.py index d68cf4c..5465f83 100644 --- a/tests/test_privacy_filter.py +++ b/tests/test_privacy_filter.py @@ -388,6 +388,44 @@ def fake_model_filter(session, pf_config): assert seen == [True] +def test_redaction_policy_version_changes_with_policy(): + base = _exp._PrivacyFilterConfig(enabled=False) + v0 = _exp.redaction_policy_version([], [], base) + # Same inputs -> same version (deterministic). + assert v0 == _exp.redaction_policy_version([], [], base) + # Adding a redact string changes the version (policy tightened). + assert _exp.redaction_policy_version(["AcmeCorp"], [], base) != v0 + # Enabling the model filter changes the version. + assert _exp.redaction_policy_version([], [], _exp._PrivacyFilterConfig(enabled=True)) != v0 + + +def test_carry_forward_skips_when_stamp_current(monkeypatch): + # A record already stamped with the CURRENT policy version must be returned + # verbatim without re-running redaction (the keystone scale optimization). + monkeypatch.setattr(_exp, "_read_privacy_filter_config", lambda: _exp._PrivacyFilterConfig(enabled=False)) + + redaction = {"redact_strings": [], "redact_usernames": []} + version = _exp.redaction_policy_version([], [], _exp._PrivacyFilterConfig(enabled=False)) + + transform_calls = [] + monkeypatch.setattr( + _exp, "transform_session", + lambda *a, **k: transform_calls.append(1) or (a[0], 0), + ) + + redact_fn = _exp._build_carry_forward_redactor(redaction) + + current = {"source": "claude", "session_id": "r1", "redaction_policy": version, "messages": []} + out = redact_fn(current) + assert out is current # untouched + assert transform_calls == [] # transform_session NOT called + + stale = {"source": "claude", "session_id": "r2", "redaction_policy": "OLD", "messages": []} + redact_fn(stale) + assert transform_calls == [1] # stale record IS re-redacted + assert stale["redaction_policy"] == version # and re-stamped + + def _match_pipe(entity: str, needle: str): def pipe(text): start = text.find(needle) From a4ff9f7b400077d4e17c1ba7a9300d88e21f2494 Mon Sep 17 00:00:00 2001 From: POM Date: Sat, 30 May 2026 02:25:03 +0200 Subject: [PATCH 5/5] docs: record merge-review findings and the O(new-data) fixes Co-Authored-By: Claude Opus 4.8 (1M context) --- ...liability-merge-and-privacy-filter-plan.md | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/docs/reliability-merge-and-privacy-filter-plan.md b/docs/reliability-merge-and-privacy-filter-plan.md index e560109..c2efd8c 100644 --- a/docs/reliability-merge-and-privacy-filter-plan.md +++ b/docs/reliability-merge-and-privacy-filter-plan.md @@ -160,6 +160,32 @@ Bugs / gaps (line refs on the `5d0a741` version): and **harden secret patterns + salt the anonymizer**. (From the first audit.) 5. Real cross-process run lock; batch the 3 HF uploads into one `create_commit`. +## Merge-quality review (4-perspective sense-check) + fixes shipped + +A focused review of the update/merge mechanism (correctness, reliability, elegance, scale) +found the design **fundamentally right and reliable** (fail-closed, convergent, self-healing, +optimistic-concurrency correct) but flagged a scale problem the carry-forward re-redaction fix +introduced. Shipped in `619ef63`: + +- **Keystone — redaction-policy stamp.** Each record is stamped with + `redaction_policy_version()` (hash of redact strings/usernames + model config + code + version). The carry-forward redactor skips re-redaction for records already stamped current, + re-redacting only stale ones. Turns a steady-state push from **O(total history)** (was + re-running the PII model over the whole dataset every push) into **~O(new data)**, while + preserving tighten-only (policy change bumps the version → one-time full re-scan). +- **Skip-unchanged.** If the merged file is byte-identical to the downloaded remote, skip all + uploads — no more empty-commit churn from the metadata timestamp. +- **Corrupt-line guard.** `_load_raw_records` preserves unparseable JSONL lines verbatim + instead of raising — a corrupt remote line can't drop data or wedge all future pushes. + +### Known/deferred from the merge review (not yet done) +- **Memory**: merge materializes both full files in RAM → OOM risk past ~1–2 GB (streaming merge). +- **Network**: full re-download + re-upload every push (sharding — larger change). +- **Compaction semantics**: "more messages wins" keeps the stale pre-compaction copy when a + conversation was legitimately summarized (keeps data, doesn't lose it; judgment call). +- **Atomicity**: 3 uploads are separate commits (self-heals; `create_commit` would make atomic). +- Model-load-status publish-time signal; `git_branch` redaction; dead UI telemetry. (From earlier audits.) + ## Open verification items (couldn't run live) - `huggingface_hub` version: confirm `upload_file(parent_commit=...)` support and that conflicts surface as HTTP 412; confirm `hf_hub_download` error classes.