diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index e1676dd..626cc3e 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -6,20 +6,64 @@ on: - "v*" permissions: - contents: write + contents: read jobs: + verify: + runs-on: ubuntu-latest + strategy: + matrix: + python-version: ["3.9", "3.12"] + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + - name: Install package + run: python -m pip install --no-deps . + - name: Compile and test + run: | + python -m compileall -q src audit-local-files/scripts tests + for test in tests/*_test.py; do python "$test"; done + release: + needs: verify runs-on: ubuntu-latest + permissions: + contents: write steps: - uses: actions/checkout@v4 - uses: actions/setup-python@v5 with: python-version: "3.12" + - name: Verify tag and package version + env: + RELEASE_TAG: ${{ github.ref_name }} + run: | + python - <<'PY' + import os + import tomllib + from pathlib import Path + + version = tomllib.loads(Path("pyproject.toml").read_text(encoding="utf-8"))["project"]["version"] + expected = f"v{version}" + if os.environ["RELEASE_TAG"] != expected: + raise SystemExit(f"release tag {os.environ['RELEASE_TAG']!r} must equal {expected!r}") + PY - name: Build distributions run: | - python -m pip install build + python -m pip install build twine python -m build + python -m twine check dist/* + python - <<'PY' + import zipfile + from pathlib import Path + + wheel = next(Path("dist").glob("*.whl")) + with zipfile.ZipFile(wheel) as archive: + names = set(archive.namelist()) + assert "clean_your_data/web/index.html" in names + PY - name: Create GitHub release env: GH_TOKEN: ${{ github.token }} diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index e368bc0..ba3141c 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -31,6 +31,10 @@ jobs: run: python tests/interactive_test.py - name: Run terminal UI test run: python tests/tui_test.py + - name: Run local GUI test + run: python tests/gui_test.py + - name: Run Agent trace test + run: python tests/trace_test.py - name: Run cleanup loop test run: python tests/cleanup_test.py - name: Run workspace actions test diff --git a/CHANGELOG.md b/CHANGELOG.md index b3ba12b..b6ff5fd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,24 @@ # Changelog +## 0.4.0 - 2026-08-12 + +- Added `cyd gui [PATH]`, a local browser interface backed by the same scanner and cleanup gates as the TUI. +- Added expandable relative-size browsing, bounded local previews, search, sorting, relationships, per-path Agent conversations, and responsive desktop/mobile layouts. +- Added secret-free AI configuration for Codex, custom stdin/stdout commands, or AI-off operation. +- Added exact-path cleanup baskets, system Trash moves, rescanning, and undo to the GUI. +- Protected active roots, VCS metadata, credential stores, and common credential files from cleanup. +- Bound the GUI to loopback with Host validation, per-run API tokens, cross-origin checks, path redaction, restrictive response headers, and no traceback disclosure. +- Added the opt-in `cyd trace -- ` wrapper for Agent sessions. +- Records metadata-only created, modified, and deleted paths while the traced command runs. +- Persists local session and event evidence in `~/.clean-your-data/provenance.sqlite3`. +- Reports attribution as an observed association, not as kernel-level proof of the exact writer process. + +## 0.3.1 - 2026-08-10 + +- Added relative size bars to the space map so large siblings stand out immediately. +- Added a first-screen space summary with folder, file, rebuildable, and staged counts. +- Reworked the folder inspector into a plain-language space story with a next-best action. + ## 0.3.0 - 2026-08-10 - Added the installable `clean-your-data` Python package. diff --git a/MANIFEST.in b/MANIFEST.in index eae7b6e..b15c445 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -3,3 +3,4 @@ include README.md include CHANGELOG.md recursive-include audit-local-files *.md *.yaml recursive-include audit-local-files/scripts *.py +recursive-include src/clean_your_data/web *.html diff --git a/PRIVACY.md b/PRIVACY.md index a3675e7..2b5a9af 100644 --- a/PRIVACY.md +++ b/PRIVACY.md @@ -1,6 +1,14 @@ # Privacy -Clean Your Data is designed for local, metadata-first analysis. The bundled scanner does not make network requests and does not upload reports. An agent may download this public repository, but the scan itself runs on the user's machine. The optional TUI cleanup loop can move an exact user-confirmed path to the local system Trash; that action and its undo record stay on the machine. +Clean Your Data is designed for local, metadata-first analysis. The bundled scanner and GUI server do not make network requests and do not upload reports. An agent may download this public repository, but the scan itself runs on the user's machine. The optional GUI/TUI cleanup loop can move an exact user-confirmed path to the local system Trash; that action and its undo record stay on the machine. + +## Local Browser GUI + +- The GUI binds only to `127.0.0.1` on a random port by default. +- It validates the loopback Host; its JSON API requires a random per-run token and rejects cross-origin browser requests. +- Public API responses omit absolute local paths and internal `_local_path` fields. +- The browser receives redacted paths, bounded metadata, and only the preview the user explicitly selects. +- Closing the server ends the session. Agent conversations in the browser are not persisted by Clean Your Data. ## Default Redaction @@ -35,6 +43,14 @@ Avoid sharing reports created with: - Secrets, tokens, keychains, or credentials. - Source file contents. +## Optional AI Commands + +AI is disabled or configured by the user. Clean Your Data can call an already authenticated Codex CLI or a trusted custom stdin/stdout command. It stores the selected mode and custom-command arguments verbatim in `~/.clean-your-data/ai-config.json`, with user-only file permissions where supported. The GUI reports that a command is configured but does not echo its arguments through the local browser API; `cyd config ai --show` is the explicit terminal view. There is no API-key field: never put credentials in command arguments, and keep them in the provider's environment or credential store. + +The opt-in tracer also stores the traced command and its arguments verbatim in the local `~/.clean-your-data/provenance.sqlite3` database. Do not put credentials directly in a traced command line. The state directory and database use user-only permissions where the operating system supports them. + +The stdin prompt constructed by Clean Your Data contains a redacted path, name, kind, size, modified time, category, and measurement status. Clean Your Data does not place the selected file preview or file contents in that prompt. The built-in Codex mode uses a read-only, ephemeral sandbox. A custom command still has its own operating-system permissions and may use its own network connection or credential store; configure only a trusted command and review that provider's privacy behavior separately. + ## Sharing Reports Before sharing a report publicly: diff --git a/README.md b/README.md index 5f05079..40142c5 100644 --- a/README.md +++ b/README.md @@ -8,48 +8,100 @@ Clean Your Data: a safe map before you clean

-**A local-first file system interface for humans and AI agents.** +**A local-first disk explorer that helps humans and Agents understand a path before anything moves.** -Browse any path, understand what lives there, and take reversible actions. +**Clean Your Data** turns an opaque folder into an interactive space map. Open a directory, follow the largest branches, preview a file locally, ask your own Agent about one exact path, and move only reviewed items to system Trash with undo. -**Clean Your Data** is a keyboard-first terminal disk explorer. It turns an opaque folder into a navigable map with search, filters, sorting, tabs, file previews, local AI questions, duplicate detection, and approval-gated moves to Trash. It is built for humans first, with a metadata contract that local Agents can understand. +The browser GUI and keyboard-first TUI use the same scanner and safety gates. AI is optional and user-configured; the explorer works without an API key or network connection.

- Real terminal demo of Clean Your Data + Clean Your Data local browser GUI

-

- Watch or download the 20-second MP4 demo -

- -## Install And Explore +## Install And Open -Install the package and launch the explorer from any directory: +Install directly from GitHub, then open any directory: ```bash -# Install the current GitHub version immediately. uv tool install git+https://github.com/MicroMilo/clean-your-data.git -# Explore the current directory, or pass any path you want to inspect. -cyd +# Local browser GUI +cyd gui ~/Documents/project + +# Terminal UI cyd ~/Documents/project ``` -When the package is published to PyPI, the install becomes: +From PyPI after the first package release: + +```bash +uv tool install clean-your-data +``` + +`cyd gui` starts a random-port server on `127.0.0.1` and opens the browser. It is not exposed to the LAN. Stop it from the square button or with `Ctrl-C`. For a source checkout, run `python3 -m pip install .` first. + +## Why This Exists + +| Question | Product behavior | +| --- | --- | +| Where did the space go? | A relative-size tree makes large siblings visible and loads deeper folders on demand. | +| What is this path? | Local preview, metadata, inferred relationships, and a per-path Agent conversation stay together. | +| Can I move it? | The scanner separates rebuildable candidates, review items, and protected data. A suggestion is never permission. | +| What if the decision is wrong? | Cleanup stages an exact path, confirms it again, moves it to system Trash, rescans, and supports undo. | +| Can I stay in the terminal? | The TUI provides the same scan, preview, Agent context, cleanup gate, Trash, and undo model. | + +This is not a generic file manager and it does not promise automatic deletion. It is the investigation and decision layer between an unfamiliar path and a file operation. + +## Optional Agent + +The explorer does not bundle an LLM account. It can use an already authenticated Codex CLI or any trusted stdin/stdout command: ```bash -python3 -m pip install --user clean-your-data -# or: uv tool install clean-your-data +cyd config ai --auto +cyd config ai --codex +cyd config ai --command 'ollama run qwen3:8b' +cyd config ai --off +cyd config ai --show ``` -For a checkout under development, run `python3 -m pip install .`. `cyd --version` prints the installed version. The demo uses a sanitized copy of this repository; it does not use a real user's home directory, `~/github`, or `node_modules`. +The GUI exposes the same setting. Commands are executed as direct arguments, never through a shell. Clean Your Data has no API-key field, but saved custom-command arguments are stored verbatim. Never put a credential in the command; keep provider credentials in that provider's own environment or credential store. + +The prompt that Clean Your Data writes to the configured command contains only bounded metadata for the selected path: redacted path, name, kind, size, modified time, category, and measurement status. Clean Your Data does not put file previews or file contents in that prompt. The built-in Codex mode runs in a read-only, ephemeral sandbox; a custom command still has its own operating-system permissions, so configure only a command you trust. Agent answers are advice and cannot approve or execute cleanup through Clean Your Data. + +## Optional Agent Trace + +Run an Agent through `cyd trace` when you want to know which paths changed during that session: + +```bash +# Trace the current directory while Codex works. +cyd trace -- codex + +# Trace a specific project while Claude Code works there. +cyd trace --path ~/projects/demo -- claude + +# Watch more than one local scope and keep a machine-readable report. +cyd trace \ + --path ~/projects/demo \ + --path ~/.codex \ + --format json \ + --output trace.json \ + -- codex +``` + +The command after `--` runs normally. The tracer takes bounded metadata snapshots while it runs and reports paths that were created, modified, deleted, or observed briefly. It records the session, command, process id, time, scope, and before/after stat fields, but never reads file contents or environment variables. Trace records stay in the local `~/.clean-your-data/provenance.sqlite3` database with user-only permissions where supported. Command arguments are stored verbatim, so never put credentials directly on a traced command line. The default scope is the command's working directory; use `--path` explicitly when an Agent writes elsewhere. + +The report says that a change was **observed while the traced command was running**. It does not claim kernel-level proof of the exact child process that wrote a path. Historical files that were created before tracing may remain unattributed. This is intentional: evidence, inference, and unknowns stay separate. ## Privacy Boundary - Read-only by default. The initial map reads metadata, not file contents. -- No uploads. The scanner never changes files; the TUI can move only an exact, user-confirmed path to the system Trash. +- No uploads by the scanner or local GUI. A configured Agent command follows its own network policy. +- The GUI listens only on `127.0.0.1`; it validates the loopback Host, requires a random per-run API token, and rejects cross-origin browser requests. +- Cleanup is separate from scanning. GUI and TUI can move only an exact, user-confirmed, eligible path to system Trash. - Exact duplicate matching is opt-in and hashes candidate files locally; raw hashes are not written to reports. - Home paths are redacted to `~` by default. Review project and folder names before sharing a report. +- The active scope, home/root, Trash, VCS metadata, credential stores, credential config, app-managed data, cloud-sync roots, unknown paths, and incomplete measurements are blocked from cleanup. +- Agent tracing is opt-in, metadata-only, local, and bounded by the selected `--path` roots and `--max-entries` limit. The traced Agent may have its own network or authentication behavior; the tracer does not add network access. ## Give It To Your Agent @@ -80,16 +132,16 @@ The optional Skill adds a repeatable audit protocol for Agents. The package is t ## The Workflow ```text -discover -> classify -> explain -> assess risk -> plan action -> compare over time +map -> inspect -> ask -> assess risk -> stage -> confirm -> Trash -> undo ``` The useful question is not only “what is large?” It is: > What is this data, who owns it, can it be rebuilt, and what can I safely do next? -## Quick Start +## Detailed Terminal Usage -After installing the package: +After installing the package, open the TUI: ```bash cyd ~/Documents/project @@ -290,11 +342,11 @@ No third-party Python packages are required. ## 中文说明 -**先看清，再动手。** +**先看清，再动手。面向人和 Agent 的本地磁盘浏览器。** -**Clean Your Data** 把电脑里的文件沉积转换成一份隐私友好、可以做决定的本地审计报告。它会检查 Desktop、Downloads、AI 工作区、Git 仓库、微信/飞书等协作 App、本地云盘，以及 `node_modules`、`.venv`、`build` 等可重建产物。 +**Clean Your Data** 把陌生目录变成可以继续探索的空间地图。你可以沿着大目录往下看，在本地预览文件，针对某个精确路径询问自己配置的 Agent，再把经过复核的项目移入系统废纸篓并撤销。 -默认只读、只读取元数据；TUI 仅在选中文件时提供受限的本地预览，不上传数据。清理时必须先用 `dd` 加入候选，再确认精确路径；确认后只移入系统废纸篓，不做永久删除，动作完成后会重新扫描指定路径，`u` 可以撤销最近一次移动。 +GUI 和 TUI 共用同一套扫描器与安全门禁。默认只读取元数据；选中文件时才在本地读取最多 4 KB、14 行预览，疑似凭据和二进制文件不会展示。AI 完全可选，不配置也能正常浏览、分析关系和进行可逆操作。 ### 交给你的 Agent @@ -310,12 +362,12 @@ Agent 可以下载仓库、读取 `audit-local-files/SKILL.md`，然后在用户 ### 为什么不直接问 Codex -Codex 可以完成一次分析，但这个仓库把每次都应该保持一致的部分固定下来：覆盖范围、隐私边界、分类标准、风险判断、行动审批和历史比较。它不是替代 Codex，而是让 Codex 以一套可重复、可审计的本地数据体检流程工作。 +Codex 可以完成一次分析，但很难持续保留“当前选中了什么、它和空间分布的关系、预览边界、清理篮状态以及撤销记录”。这个产品把这些状态放进一个人可以直接操控的界面，并只给 Agent 一份结构化、受限的元信息。Agent 负责解释，人负责决定，程序负责执行安全门禁。 ### 工作流 ```text -发现 -> 分类 -> 解释 -> 追问 -> 判断风险 -> 生成行动计划 -> 定期比较 +空间地图 -> 检查 -> 询问 -> 判断风险 -> 加入清理篮 -> 精确确认 -> 废纸篓 -> 撤销 ``` 真正要回答的不是“哪里最大”，而是： @@ -324,17 +376,41 @@ Codex 可以完成一次分析，但这个仓库把每次都应该保持一致 ### 快速运行 -安装包后，从任意目录打开磁盘浏览器： +安装包后，从任意目录打开浏览器 GUI 或终端 TUI： ```bash # 直接安装 GitHub 当前版本 uv tool install git+https://github.com/MicroMilo/clean-your-data.git -# 浏览当前目录，或传入指定路径 -cyd +# 本地浏览器 GUI +cyd gui ~/Documents/project + +# 终端 TUI cyd ~/Documents/project ``` +`cyd gui` 只监听 `127.0.0.1` 的随机端口，不会暴露到局域网；服务端还会校验本地 Host。每次运行都有随机 API 令牌，并拒绝跨站请求。方形停止按钮和 `Ctrl-C` 都可以关闭本地服务。 + +AI 不需要内置在产品里。你可以复用本机已经登录的 Codex CLI，接入 Ollama 等可信命令，或彻底关闭： + +```bash +cyd config ai --auto +cyd config ai --codex +cyd config ai --command 'ollama run qwen3:8b' +cyd config ai --off +cyd config ai --show +``` + +配置没有 API-key 字段，也不会通过 shell 执行命令，但自定义命令及参数会原样保存在本地，因此绝不能把密钥写进命令参数。凭据应继续放在对应工具自己的环境变量或凭据存储中。Clean Your Data 写入该命令 stdin 的 Prompt 只包含所选路径的匿名路径、名称、类型、大小、修改时间、分类和测量状态，不包含右侧文件预览或文件正文。内置 Codex 模式使用只读、临时沙箱；自定义命令仍拥有它自己的操作系统权限，因此只能配置你信任的命令。 + +如果你想观察 Agent 在一个项目里留下了哪些文件，可以直接包住它运行： + +```bash +cyd trace --path ~/Documents/project -- codex +``` + +`trace` 只记录运行期间观察到的创建、修改和删除路径，以及命令、时间和前后元信息；不会读取文件正文。默认把追踪记录保存在 `~/.clean-your-data/provenance.sqlite3`，并在系统支持时设为仅当前用户可读。命令及参数会原样保存在本地，因此不要把密钥直接写进被追踪的命令行。它能证明“在这次被追踪的 Agent 会话期间观察到了变化”，不能对历史文件或并发进程做超出证据的断言。 + 开发仓库时可以在仓库根目录执行 `python3 -m pip install .`。发布到 PyPI 后，也可以执行 `python3 -m pip install --user clean-your-data`。`cyd --version` 查看版本。 旧的脚本入口仍然保留，方便已有用户和 Agent 继续使用： @@ -392,13 +468,15 @@ JSON 报告还会区分 `measured`、`timeout`、`error`、`missing` 和 `unknow ### 隐私与安全 -扫描器只读取路径、占用大小、修改时间和可选的 Git 状态计数；不会读取聊天、浏览器历史、邮件正文、文档正文、源码、凭据或 keychain。选中文件时，TUI 会额外提供一个本地预览例外，最多读取 4 KB、14 行，跳过二进制和疑似凭据文件，且不会发送给 Codex。默认把 home 目录显示为 `~`，默认不输出 Git origin URL。清理历史保存在本机 `~/.clean-your-data/cleanup-history.json`，书签、最近路径和最后工作区保存在 `~/.clean-your-data/workspace-state.json`；这些文件只在本地使用，不会写入报告或上传。任何公开分享前，都应检查项目名和文件夹名。 +扫描器只读取路径、占用大小、修改时间和可选的 Git 状态计数；不会读取聊天、浏览器历史、邮件正文、文档正文、源码、凭据或 keychain。选中文件时，GUI/TUI 会额外提供一个本地预览例外，最多读取 4 KB、14 行，跳过二进制和疑似凭据文件，且不会发送给 Agent。默认把 home 目录显示为 `~`，默认不输出 Git origin URL。清理历史保存在本机 `~/.clean-your-data/cleanup-history.json`，书签、最近路径和最后工作区保存在 `~/.clean-your-data/workspace-state.json`；这些文件只在本地使用，不会写入报告或上传。任何公开分享前，都应检查项目名和文件夹名。 + +扫描根目录、home、系统根目录、废纸篓、`.git`/`.ssh` 等关键目录、`.env` 等凭据配置、App 私有数据、云同步目录、未知分类和未完整测量的路径都不能进入清理篮。Agent 的回答只是建议，不能绕过这些确定性规则。 项目维护时会使用真实本机审计的聚合匿名证据包，让不同 Agent 角色独立评审，再通过确定性 fixture 和 schema 测试后发布；原始本机报告不会进入仓库。 ### 依赖 -安装包没有第三方运行时依赖。需要 Python 3.9+；macOS/Linux 上的 `du`；只有在进行 Git 检查时才需要 `git`；TUI 需要支持 `curses` 的交互式终端。若要按 `a` 或 `dd` 询问 Codex，还需要本机已登录的 `codex` CLI，或自行配置 `CLEAN_YOUR_DATA_AI_COMMAND`；`D` 的深度关系分析只使用 Python 标准库。`t`/`v`/`c`/`o` 会调用本机已有的 Terminal、VS Code、Cursor 或 Finder；移动到系统废纸篓使用本机文件系统。 +安装包没有第三方运行时依赖。需要 Python 3.9+；macOS/Linux 上的 `du`；只有在进行 Git 检查时才需要 `git`；TUI 需要支持 `curses` 的交互式终端；GUI 使用 Python 标准库启动本地 HTTP 服务并调用默认浏览器。需要 AI 解释时，还要有本机已登录的 `codex` CLI，或自行配置可信的 stdin/stdout 命令；其余浏览与清理功能不依赖 LLM。 ## License diff --git a/RELEASING.md b/RELEASING.md new file mode 100644 index 0000000..ccb2efb --- /dev/null +++ b/RELEASING.md @@ -0,0 +1,35 @@ +# Releasing Clean Your Data + +Releases are tag-driven. Do not create a tag until every step below passes from a clean checkout. + +## Preflight + +1. Confirm `pyproject.toml`, `src/clean_your_data/__init__.py`, and `CHANGELOG.md` use the same version. +2. Review `README.md`, `PRIVACY.md`, and `SECURITY.md` whenever data flow or cleanup behavior changes. +3. Check tracked files for personal absolute paths, credentials, local reports, databases, and cleanup history. +4. Run every test on Python 3.9 and 3.12 through GitHub Actions. + +## Local Build + +```bash +python3 -m venv /tmp/cyd-release-venv +/tmp/cyd-release-venv/bin/python -m pip install --upgrade pip build twine +/tmp/cyd-release-venv/bin/python -m compileall -q src audit-local-files/scripts tests +for test in tests/*_test.py; do /tmp/cyd-release-venv/bin/python "$test"; done +/tmp/cyd-release-venv/bin/python -m build +/tmp/cyd-release-venv/bin/python -m twine check dist/* +``` + +Install the wheel into a second empty environment and verify `cyd --version`, `cyd gui --help`, `cyd config ai --show`, and a bounded scan. Confirm the wheel contains `clean_your_data/web/index.html`. + +## Tag + +After reviewing the exact commit: + +```bash +git tag -a v0.4.0 -m "Clean Your Data v0.4.0" +git push origin main +git push origin v0.4.0 +``` + +The release workflow reruns compilation and tests, builds the wheel and source archive, validates both distributions, checks the packaged GUI asset, and attaches the artifacts to the GitHub release. diff --git a/SECURITY.md b/SECURITY.md index 5e90b38..b1538ce 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -17,5 +17,9 @@ Security-sensitive issues include: - Moving a path without the exact-path confirmation flow, or bypassing the system Trash and undo record. - Overwriting an existing path during undo. - Passing a selected path through shell interpolation instead of a direct argument when opening Terminal, VS Code, Cursor, or Finder. +- Exposing the GUI beyond loopback, accepting an unauthenticated API request, or returning internal absolute paths in a public API response. +- Storing provider API keys in the Clean Your Data AI configuration. -The scanner should remain read-only. The TUI's only mutating operation is an explicit move of one measured, eligible path into system Trash after `dd`, `Y`, and `y`; it must never permanently delete, bulk-delete, or touch app-managed, cloud-sync, unknown, home, or root paths. +The scanner should remain read-only. GUI and TUI may mutate files only through an explicit move of measured, eligible, exactly confirmed paths into system Trash. They must never permanently delete or touch the active scan root, home/root, Trash, VCS metadata, credential stores, credential configuration, app-managed data, cloud-sync roots, unknown paths, or incomplete measurements. + +The GUI must bind to `127.0.0.1`, validate the loopback Host, require a random per-run token on every API request, reject cross-origin browser requests, return no internal `_local_path`, apply a restrictive content security policy, and avoid exposing tracebacks to the browser. Custom AI commands must be parsed into direct arguments and must never be executed through a shell. diff --git a/assets/clean-your-data-gui-v0.4.jpg b/assets/clean-your-data-gui-v0.4.jpg new file mode 100644 index 0000000..f991d6b Binary files /dev/null and b/assets/clean-your-data-gui-v0.4.jpg differ diff --git a/audit-local-files/SKILL.md b/audit-local-files/SKILL.md index 165c815..0a5b8a5 100644 --- a/audit-local-files/SKILL.md +++ b/audit-local-files/SKILL.md @@ -1,6 +1,6 @@ --- name: audit-local-files -description: Turn local file sprawl into a privacy-preserving, decision-ready audit. Use when someone asks to analyze or organize Desktop, Downloads, Documents, Codex or other AI workspaces, Git repositories, cloud-sync folders, or local storage from Feishu/Lark, WeChat, Slack, Teams, Discord, Telegram, QQ, DingTalk, Zoom, mail, or browsers. Also use when proposing a safe archive, migration, cleanup, or before/after storage comparison. +description: Explain local file provenance and turn local file sprawl into a privacy-preserving, decision-ready audit. Use when someone asks what an Agent created, which paths changed during a local Agent session, what owns a file, or how to analyze Desktop, Downloads, Documents, Codex or other AI workspaces, Git repositories, cloud-sync folders, or local storage from Feishu/Lark, WeChat, Slack, Teams, Discord, Telegram, QQ, DingTalk, Zoom, mail, or browsers. Also use when proposing a safe archive, migration, cleanup, or before/after storage comparison. --- # Clean Your Data @@ -19,6 +19,7 @@ Use this skill as a local data audit protocol around the `cyd` terminal file-sys - When the user names a path for closer inspection, keep that path redacted too. The interactive map may expose only home-relative paths by default. - Exact duplicate detection is opt-in. Only run `--duplicates` after the user explicitly asks for duplicate matching; it reads candidate file bytes locally for SHA-256, never uploads them, and never changes files. - Duplicate groups are review evidence, not cleanup authorization. Preserve the distinction between independent copies and hard-link aliases, and do not add duplicate bytes to parent target or artifact totals. +- Agent tracing is opt-in and bounded. It records metadata-only changes under explicit `--path` roots while a user-selected command runs; it does not read contents or environment variables, and it does not claim kernel-level proof of the exact child process. ## Workflow @@ -72,6 +73,19 @@ The map is metadata-first. It lets the user select a folder or file, inspect its The advanced browsing controls are `/` fuzzy search, `f` filter, `s` sort, `T` local tags, `N` new tab, `gt`/`gT` next or previous tab, and `X` close tab. Search and filters operate only on nodes already loaded into the map; they do not imply that an unopened directory was scanned. Tags are stored locally with the existing workspace state and are not included in reports or Agent prompts. +### Agent Session Trace + +When the user wants to know what an Agent changed during a session, run the Agent through the package tracer instead of trying to reconstruct causality from old timestamps: + +```bash +cyd trace --path /path/to/project -- codex +cyd trace --path /path/to/project -- claude +``` + +The command after `--` runs normally. The tracer takes bounded metadata snapshots while it runs and reports `created`, `modified`, `deleted`, and transient `created + deleted` paths. It stores the local session and event evidence in `~/.clean-your-data/provenance.sqlite3`. Use repeated `--path` options when the Agent writes to more than one known scope, and `--format json --output trace.json` for a structured handoff. + +Interpret the result as: "this path changed while the traced command was running." Do not rewrite that as proof that a particular child process created it. If the file predates the trace session, report its origin as unknown or inferred from separate evidence. A snapshot can miss a very short-lived change, hit the `--max-entries` limit, or be unable to read a protected path; preserve those limitations in the conclusion. + ### TUI Cleanup Loop `dd` is a Vim-style operator: it stages the exact selected path and does not move anything. The right pane shows two separate layers: diff --git a/pyproject.toml b/pyproject.toml index db0e687..c6210d5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,8 +4,8 @@ build-backend = "setuptools.build_meta" [project] name = "clean-your-data" -version = "0.3.0" -description = "A keyboard-first disk explorer for humans and AI agents" +version = "0.4.0" +description = "A local-first disk explorer for humans and AI agents" readme = "README.md" requires-python = ">=3.9" license = "MIT" @@ -14,9 +14,11 @@ authors = [ { name = "Clean Your Data contributors" }, ] dependencies = [] +keywords = ["disk-usage", "file-browser", "local-first", "terminal-ui", "ai-agent"] classifiers = [ "Development Status :: 4 - Beta", "Environment :: Console", + "Environment :: Web Environment", "Intended Audience :: End Users/Desktop", "Operating System :: MacOS", "Operating System :: POSIX", @@ -41,3 +43,6 @@ package-dir = { "" = "src" } [tool.setuptools.packages.find] where = ["src"] + +[tool.setuptools.package-data] +clean_your_data = ["web/*.html"] diff --git a/src/clean_your_data/__init__.py b/src/clean_your_data/__init__.py index 7b68bb8..cee5b94 100644 --- a/src/clean_your_data/__init__.py +++ b/src/clean_your_data/__init__.py @@ -1,3 +1,3 @@ """Clean Your Data: a local-first terminal file explorer.""" -__version__ = "0.3.0" +__version__ = "0.4.0" diff --git a/src/clean_your_data/ai_config.py b/src/clean_your_data/ai_config.py new file mode 100644 index 0000000..ca00de0 --- /dev/null +++ b/src/clean_your_data/ai_config.py @@ -0,0 +1,218 @@ +"""Local, secret-free configuration for optional AI commands.""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import shutil +from pathlib import Path +from typing import Any, Optional + + +STATE_ENV = "CLEAN_YOUR_DATA_STATE_DIR" +AI_COMMAND_ENV = "CLEAN_YOUR_DATA_AI_COMMAND" +AI_CONFIG_FILE = "ai-config.json" +VALID_AI_MODES = {"auto", "codex", "command", "off"} + + +def state_dir() -> Path: + configured = os.environ.get(STATE_ENV, "").strip() + return Path(configured).expanduser() if configured else Path.home() / ".clean-your-data" + + +def ai_config_path() -> Path: + return state_dir() / AI_CONFIG_FILE + + +def default_ai_config() -> dict[str, Any]: + return {"version": 1, "mode": "auto", "command": []} + + +def normalize_ai_config(data: Any) -> dict[str, Any]: + config = default_ai_config() + if not isinstance(data, dict): + return config + mode = str(data.get("mode") or "auto").strip().lower() + if mode not in VALID_AI_MODES: + mode = "auto" + command_value = data.get("command") + if isinstance(command_value, str): + try: + command = shlex.split(command_value) + except ValueError: + command = [] + elif isinstance(command_value, list): + command = [str(item) for item in command_value if str(item)] + else: + command = [] + if mode == "command" and not command: + mode = "off" + return {"version": 1, "mode": mode, "command": command[:64]} + + +def load_ai_config(path: Optional[Path] = None) -> dict[str, Any]: + target = path or ai_config_path() + try: + data = json.loads(target.read_text(encoding="utf-8")) + except (FileNotFoundError, OSError, json.JSONDecodeError): + return default_ai_config() + return normalize_ai_config(data) + + +def save_ai_config(config: dict[str, Any], path: Optional[Path] = None) -> dict[str, Any]: + target = path or ai_config_path() + normalized = normalize_ai_config(config) + target.parent.mkdir(parents=True, exist_ok=True) + try: + os.chmod(target.parent, 0o700) + except OSError: + pass + temporary = target.with_suffix(target.suffix + ".tmp") + temporary.write_text(json.dumps(normalized, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + try: + os.chmod(temporary, 0o600) + except OSError: + pass + os.replace(temporary, target) + return normalized + + +def parse_command(value: str) -> list[str]: + if len(value) > 4096: + raise ValueError("AI command is too long") + try: + command = shlex.split(value) + except ValueError as exc: + raise ValueError(f"invalid AI command: {exc}") from exc + if not command: + raise ValueError("AI command cannot be empty") + if len(command) > 64: + raise ValueError("AI command has too many arguments") + return command + + +def config_for_display( + config: Optional[dict[str, Any]] = None, + *, + reveal_command: bool = False, +) -> dict[str, Any]: + configured_env = os.environ.get(AI_COMMAND_ENV, "").strip() + if configured_env: + try: + command = parse_command(configured_env) + provider = f"Environment command ({Path(command[0]).name})" + except ValueError: + command = [] + provider = "Invalid environment command" + return { + "mode": "environment", + "command": shlex.join(command) if reveal_command and command else "", + "command_configured": bool(command), + "provider": provider, + "has_api_key_field": False, + "stores_command_arguments": False, + "managed_by_environment": True, + } + normalized = normalize_ai_config(config or load_ai_config()) + command = normalized["command"] + return { + "mode": normalized["mode"], + "command": shlex.join(command) if reveal_command and command else "", + "command_configured": bool(command), + "provider": provider_label(normalized), + "has_api_key_field": False, + "stores_command_arguments": bool(command), + "managed_by_environment": False, + } + + +def provider_label(config: Optional[dict[str, Any]] = None) -> str: + normalized = normalize_ai_config(config or load_ai_config()) + mode = normalized["mode"] + if mode == "off": + return "AI disabled" + if mode == "command": + return f"Custom command ({Path(normalized['command'][0]).name})" + if mode == "codex": + return "Codex" + return "Codex (auto)" if shutil.which("codex") else "No local AI command" + + +def resolve_ai_command(config_path: Optional[Path] = None) -> tuple[Optional[list[str]], str]: + """Resolve a direct argv command; never invoke a shell.""" + configured_env = os.environ.get(AI_COMMAND_ENV, "").strip() + if configured_env: + try: + command = parse_command(configured_env) + except ValueError: + return None, "Configured local AI command is invalid." + return command, "Configured local AI" + + config = load_ai_config(config_path) + mode = config["mode"] + if mode == "off": + return None, "AI disabled." + if mode == "command": + return list(config["command"]), provider_label(config) + + codex = shutil.which("codex") + if codex: + return ( + [ + codex, + "exec", + "--sandbox", + "read-only", + "--ephemeral", + "--skip-git-repo-check", + "--color", + "never", + "-C", + "/tmp", + "-", + ], + "Codex", + ) + if mode == "codex": + return None, "Codex CLI was not found." + return None, "No local AI command." + + +def config_main(argv: list[str]) -> int: + parser = argparse.ArgumentParser( + prog="cyd config ai", + description="Configure the optional local AI command without storing API keys.", + ) + group = parser.add_mutually_exclusive_group() + group.add_argument("--auto", action="store_true", help="Auto-detect a local Codex CLI.") + group.add_argument("--codex", action="store_true", help="Require the local Codex CLI.") + group.add_argument("--off", action="store_true", help="Disable AI while keeping the explorer available.") + group.add_argument("--command", help="Direct argv command that reads a prompt from stdin and writes an answer to stdout.") + group.add_argument("--show", action="store_true", help="Show the current secret-free configuration.") + args = parser.parse_args(argv) + + if args.command is not None: + try: + command = parse_command(args.command) + except ValueError as exc: + parser.error(str(exc)) + config = save_ai_config({"version": 1, "mode": "command", "command": command}) + elif args.codex: + config = save_ai_config({"version": 1, "mode": "codex", "command": []}) + elif args.off: + config = save_ai_config({"version": 1, "mode": "off", "command": []}) + elif args.auto: + config = save_ai_config({"version": 1, "mode": "auto", "command": []}) + else: + config = load_ai_config() + + display = config_for_display(config, reveal_command=True) + print(f"AI mode: {display['mode']}") + print(f"Provider: {display['provider']}") + if display["command"]: + print(f"Command: {display['command']}") + print("Dedicated API-key field: no") + print("Saved custom-command arguments are stored verbatim; do not include secrets.") + return 0 diff --git a/src/clean_your_data/audit_local_files.py b/src/clean_your_data/audit_local_files.py index 75ef2bf..209dd73 100644 --- a/src/clean_your_data/audit_local_files.py +++ b/src/clean_your_data/audit_local_files.py @@ -1386,6 +1386,12 @@ def add_node( # Private in-memory context for the local TUI. It is never emitted # by JSON/Markdown output and is needed for lazy expansion/preview. node["_local_path"] = str(path) + node["_stat_device"] = int(path_stat.st_dev) + node["_stat_inode"] = int(path_stat.st_ino) + node["_stat_mode"] = int(path_stat.st_mode) + node["_stat_ctime_ns"] = int( + getattr(path_stat, "st_ctime_ns", int(path_stat.st_ctime * 1_000_000_000)) + ) nodes.append(node) may_descend = not should_skip_dir(path) or (allow_skipped_root and depth == 0) if depth >= max_depth or not node["can_expand"] or not may_descend: diff --git a/src/clean_your_data/audit_tui.py b/src/clean_your_data/audit_tui.py index 669b1c6..e7bcea8 100644 --- a/src/clean_your_data/audit_tui.py +++ b/src/clean_your_data/audit_tui.py @@ -7,16 +7,19 @@ import json import os import queue -import shlex import shutil +import stat as stat_module import subprocess import sys import threading import textwrap import time +import uuid from pathlib import Path from typing import Any, Callable, Optional +from .ai_config import resolve_ai_command as resolve_configured_ai_command + DEFAULT_QUESTION = "What is this area likely used for, and what should I check before changing it?" INITIAL_TREE_LEVELS = 3 @@ -29,6 +32,9 @@ WORKSPACE_STATE_FILE = "workspace-state.json" LARGE_FILTER_BYTES = 100 * 1024 * 1024 RECENT_FILTER_SECONDS = 7 * 24 * 60 * 60 +REBUILDABLE_NAMES = {"node_modules", ".venv", "venv", "build", "dist", ".next", "target", "__pycache__"} +PROTECTED_CLEANUP_PARTS = {".git", ".hg", ".svn", ".ssh", ".gnupg", ".aws", ".kube"} +PROTECTED_CLEANUP_FILES = {".env", ".env.local", ".netrc", ".npmrc", ".pypirc"} FILTER_OPTIONS = ( ("all", "All visible areas"), ("folders", "Folders only"), @@ -130,6 +136,39 @@ def display_size(node: dict[str, Any]) -> str: return str(node.get("human_size") or "Not available") +def node_bytes(node: dict[str, Any]) -> int: + try: + return max(0, int(node.get("allocated_bytes") or 0)) + except (TypeError, ValueError): + return 0 + + +def is_rebuildable_node(node: dict[str, Any]) -> bool: + return node.get("category") == "cache" or str(node.get("name") or "") in REBUILDABLE_NAMES + + +def size_bar(node: dict[str, Any], siblings: list[dict[str, Any]], width: int = 9) -> str: + """Render a compact relative-size signal for a tree row.""" + width = max(1, width) + largest = max((node_bytes(item) for item in siblings), default=0) + value = node_bytes(node) + if largest <= 0 or value <= 0: + return "." * width + filled = max(1, round(width * value / largest)) + return "#" * min(width, filled) + "." * max(0, width - filled) + + +def space_summary(state: Any) -> str: + nodes = list(getattr(state, "nodes", []) or []) + root = next((node for node in nodes if node.get("parent_id") is None), None) + folders = sum(1 for node in nodes if node.get("kind") == "folder") + files = sum(1 for node in nodes if node.get("kind") == "file") + rebuildable = sum(1 for node in nodes if is_rebuildable_node(node)) + staged = len(getattr(state, "cleanup_queue", {}) or {}) + total = display_size(root) if root else "Not available" + return f"SUMMARY {total} total | {folders} folders | {files} files | {rebuildable} rebuildable | {staged} staged" + + def display_node_name(node: dict[str, Any]) -> str: name = str(node.get("name") or node.get("path") or "unnamed") return f"{name}/" if node.get("kind") == "folder" and not name.endswith("/") else name @@ -153,9 +192,18 @@ def sensitive_preview_path(path: Path) -> bool: "id_rsa", "id_ed25519", "known_hosts", + ".netrc", + ".npmrc", + ".pypirc", } sensitive_suffixes = (".pem", ".key", ".p12", ".pfx", ".kdbx") - return name in sensitive_names or name.endswith(sensitive_suffixes) + protected_parents = {".git", ".hg", ".svn", ".ssh", ".gnupg", ".aws", ".kube"} + return ( + name in sensitive_names + or name.startswith(".env.") + or name.endswith(sensitive_suffixes) + or any(part.lower() in protected_parents for part in path.parts) + ) def read_file_preview(node: dict[str, Any]) -> list[str]: @@ -166,11 +214,24 @@ def read_file_preview(node: dict[str, Any]) -> list[str]: path = Path(str(local_path)) if sensitive_preview_path(path): return ["Preview hidden because this filename may contain credentials or private configuration."] + descriptor: Optional[int] = None try: - with path.open("rb") as handle: - data = handle.read(PREVIEW_MAX_BYTES + 1) - except OSError as exc: - return [f"Preview unavailable: {exc}"] + flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) + descriptor = os.open(path, flags) + current_stat = os.fstat(descriptor) + if not matches_scanned_identity(node, current_stat): + return ["Preview unavailable because this file changed after the scan. Rescan before opening it."] + if not stat_module.S_ISREG(current_stat.st_mode): + return ["Preview unavailable because this path is not a regular file."] + data = os.read(descriptor, PREVIEW_MAX_BYTES + 1) + except OSError: + return ["Preview unavailable because this path could not be read."] + finally: + if descriptor is not None: + try: + os.close(descriptor) + except OSError: + pass if b"\x00" in data[:PREVIEW_MAX_BYTES]: return ["Binary file; text preview is unavailable."] truncated = len(data) > PREVIEW_MAX_BYTES @@ -210,11 +271,31 @@ def local_node_path(node: dict[str, Any]) -> Optional[Path]: return Path(str(local_path)).expanduser() +def matches_scanned_identity(node: dict[str, Any], current_stat: os.stat_result) -> bool: + """Reject object replacement while accepting older nodes without ctime metadata.""" + expected_basic = (node.get("_stat_device"), node.get("_stat_inode"), node.get("_stat_mode")) + if not all(value is not None for value in expected_basic): + return True + try: + current_basic = (int(current_stat.st_dev), int(current_stat.st_ino), int(current_stat.st_mode)) + if current_basic != tuple(int(value) for value in expected_basic): + return False + expected_ctime = node.get("_stat_ctime_ns") + if expected_ctime is None: + return True + current_ctime = int( + getattr(current_stat, "st_ctime_ns", int(current_stat.st_ctime * 1_000_000_000)) + ) + return current_ctime == int(expected_ctime) + except (TypeError, ValueError): + return False + + def path_is_within(path: Path, parent: Path) -> bool: try: path.resolve().relative_to(parent.resolve()) return True - except ValueError: + except (OSError, RuntimeError, ValueError): return False @@ -246,14 +327,26 @@ def cleanup_gate(node: dict[str, Any]) -> tuple[bool, str]: return False, "the local path is unavailable in this TUI session" try: resolved = path.resolve() - except OSError as exc: - return False, f"the path could not be resolved ({exc})" + except (OSError, RuntimeError): + return False, "the path could not be resolved" if not resolved.exists(): return False, "the path no longer exists" + try: + current_stat = path.lstat() + except OSError: + return False, "the path could not be inspected again" + if not matches_scanned_identity(node, current_stat): + return False, "the path changed since it was scanned; rescan before cleanup" if resolved == Path("/") or resolved == Path.home().resolve(): return False, "the filesystem root and home directory are protected" + if node.get("kind") == "folder" and node.get("parent_id") is None: + return False, "the active explorer scope is protected" if resolved == (Path.home() / ".Trash").resolve() or path_is_within(resolved, Path.home() / ".Trash"): return False, "the system Trash is protected" + if any(part in PROTECTED_CLEANUP_PARTS for part in resolved.parts): + return False, "version-control or credential storage is protected" + if resolved.name in PROTECTED_CLEANUP_FILES or resolved.name.startswith(".env."): + return False, "credential configuration is protected" if str(node.get("measurement_status") or "unknown") != "measured": return False, "the initial scan did not measure this path completely" category = str(node.get("category") or "unknown") @@ -326,6 +419,10 @@ def load_cleanup_history(history_path: Optional[Path] = None) -> list[dict[str, def save_cleanup_history(history: list[dict[str, Any]], history_path: Optional[Path] = None) -> None: path = history_path or cleanup_history_path() path.parent.mkdir(parents=True, exist_ok=True) + try: + os.chmod(path.parent, 0o700) + except OSError: + pass temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(history, ensure_ascii=False, indent=2), encoding="utf-8") try: @@ -367,6 +464,11 @@ def move_to_trash( source = path.expanduser() if not source.exists(): raise FileNotFoundError(f"path no longer exists: {source}") + if source.is_symlink(): + raise ValueError("symbolic links cannot be moved through the cleanup workflow") + current_stat = source.lstat() + if node and not matches_scanned_identity(node, current_stat): + raise ValueError("the path changed since it was scanned; rescan before cleanup") resolved = source.resolve() if resolved == Path("/") or resolved == Path.home().resolve(): raise ValueError("the filesystem root and home directory cannot be moved to Trash") @@ -377,7 +479,7 @@ def move_to_trash( destination = unique_trash_destination(root, source) shutil.move(str(source), str(destination)) record = { - "record_id": f"cleanup-{int(time.time() * 1000)}", + "record_id": f"cleanup-{time.time_ns()}-{uuid.uuid4().hex[:8]}", "original_path": str(resolved), "trash_path": str(destination.resolve()), "name": source.name, @@ -421,7 +523,16 @@ def restore_trash_record(record: dict[str, Any], history_path: Optional[Path] = break else: history.append(record) - save_cleanup_history(history, history_path) + try: + save_cleanup_history(history, history_path) + except OSError as exc: + try: + shutil.move(str(original), str(destination)) + except OSError as rollback_exc: + raise RuntimeError( + f"restored the Trash item but could not save history ({exc}); rollback failed ({rollback_exc})" + ) from exc + raise RuntimeError(f"could not save restore history; the restore was rolled back ({exc})") from exc return record @@ -461,6 +572,10 @@ def load_workspace_state(state_path: Optional[Path] = None) -> dict[str, Any]: def save_workspace_state(state: dict[str, Any], state_path: Optional[Path] = None) -> None: path = state_path or workspace_state_path() path.parent.mkdir(parents=True, exist_ok=True) + try: + os.chmod(path.parent, 0o700) + except OSError: + pass temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") try: @@ -649,34 +764,9 @@ def analyze_path_relationships( } -def resolve_ai_command() -> tuple[Optional[list[str]], str]: - """Prefer an explicit command, otherwise use a read-only ephemeral Codex CLI.""" - configured = os.environ.get("CLEAN_YOUR_DATA_AI_COMMAND", "").strip() - if configured: - try: - command = shlex.split(configured) - except ValueError: - return None, "Configured local AI command is invalid." - return (command or None), "Configured local AI" - codex = shutil.which("codex") - if codex: - return ( - [ - codex, - "exec", - "--sandbox", - "read-only", - "--ephemeral", - "--skip-git-repo-check", - "--color", - "never", - "-C", - "/tmp", - "-", - ], - "Codex", - ) - return None, "No local AI command" +def resolve_ai_command(config_path: Optional[Path] = None) -> tuple[Optional[list[str]], str]: + """Use the shared, shell-free AI provider configuration.""" + return resolve_configured_ai_command(config_path) def build_prompt(node: dict[str, Any], question: str = DEFAULT_QUESTION) -> str: @@ -692,6 +782,7 @@ def build_prompt(node: dict[str, Any], question: str = DEFAULT_QUESTION) -> str: f"Size: {node.get('human_size') or 'unknown'}", f"Last changed: {display_date(node.get('modified_at'))}", f"Area: {node.get('area') or 'unknown'}", + f"Category: {node.get('category') or 'unknown'}", f"Measurement: {node.get('measurement_status') or 'unknown'}", "", f"My question: {question.strip() or DEFAULT_QUESTION}", @@ -856,9 +947,13 @@ def copy_to_clipboard(value: str) -> str: return "Clipboard access failed. Press A to ask Codex or copy the context manually." -def ask_local_ai(prompt: str, cancel_event: Optional[threading.Event] = None) -> tuple[Optional[str], str]: +def ask_local_ai( + prompt: str, + cancel_event: Optional[threading.Event] = None, + config_path: Optional[Path] = None, +) -> tuple[Optional[str], str]: """Call the configured local AI, or a read-only ephemeral Codex CLI when available.""" - command, provider = resolve_ai_command() + command, provider = resolve_ai_command(config_path) if not command: return None, f"{provider} Press C to copy the metadata-only context." process: Optional[subprocess.Popen[str]] = None @@ -876,16 +971,15 @@ def ask_local_ai(prompt: str, cancel_event: Optional[threading.Event] = None) -> process.communicate() return None, f"{provider} timed out. Press C to copy the context." try: - stdout, stderr = process.communicate(input=prompt if not sent_input else None, timeout=0.2) + stdout, _stderr = process.communicate(input=prompt if not sent_input else None, timeout=0.2) break except subprocess.TimeoutExpired: sent_input = True continue - except (OSError, ValueError, subprocess.TimeoutExpired) as exc: - return None, f"{provider} did not answer: {exc}. Press C to copy the context." + except (OSError, ValueError, subprocess.TimeoutExpired): + return None, f"{provider} could not be started. Check its local configuration, then try again." if process.returncode != 0: - detail = stderr.strip() or f"exit code {process.returncode}" - return None, f"{provider} failed: {detail}. Press C to copy the context." + return None, f"{provider} exited with code {process.returncode}. Check that provider directly, then try again." answer = stdout.strip() if not answer: return None, f"{provider} returned no answer. Press C to copy the context." @@ -1868,12 +1962,13 @@ def draw_header(stdscr: Any, state: TuiState, width: int) -> None: _, ai_label = resolve_ai_command() if state.ai_busy: ai_label = f"{ai_label} thinking {state.spinner()}" - add_text(stdscr, 0, 0, "CLEAN YOUR DATA / TERMINAL EXPLORER", width, curses.A_BOLD | palette("title")) + add_text(stdscr, 0, 0, "CLEAN YOUR DATA / SPACE MAP", width, curses.A_BOLD | palette("title")) add_text(stdscr, 1, 0, f"Tab: {state.tab_status()} | Scope: {scope} | Map: {status} | Decision: {gate} | AI: {ai_label}", width, palette("status")) + add_text(stdscr, 2, 0, space_summary(state), width, palette("answer")) search = state.search_query or "off" - add_text(stdscr, 2, 0, f"VIEW Search: {search} | Filter: {state.filter_label()} | Sort: {state.sort_label()}", width, palette("status")) - add_text(stdscr, 3, 0, "MOVE j/k ENTER open/close / search f filter s sort T tags N new tab ? help q quit", width, palette("muted")) - add_text(stdscr, 4, 0, "-" * max(0, width), width, palette("rule")) + add_text(stdscr, 3, 0, f"VIEW Search: {search} | Filter: {state.filter_label()} | Sort: {state.sort_label()}", width, palette("status")) + add_text(stdscr, 4, 0, "MOVE j/k ENTER open / search A ask agent dd review ? help q quit", width, palette("muted")) + add_text(stdscr, 5, 0, "SPACE MAP # larger bars mean more space within this folder", width, palette("rule")) def draw_tree(stdscr: Any, state: TuiState, y: int, height: int, width: int) -> None: @@ -1900,11 +1995,49 @@ def draw_tree(stdscr: Any, state: TuiState, y: int, height: int, width: int) -> tagged = "@" if state.node_tags(node) else " " indent = " " * min(int(node.get("depth") or 0), 8) name = display_node_name(node) - line = f"{selected}{basket}{tagged} {marker} {indent}{name} {display_size(node)}" + siblings = by_parent.get(node.get("parent_id"), [node]) + bar = size_bar(node, siblings) + line = f"{selected}{basket}{tagged} {marker} {indent}{name} {bar} {display_size(node)}" attr = curses.A_BOLD | palette("selected") if selected == "*" else palette("basket") if basket == "+" else palette("folder") if node.get("kind") == "folder" else 0 add_text(stdscr, y + row, 0, line, width, attr) +def space_story_lines(state: TuiState, node: dict[str, Any], width: int) -> list[tuple[str, int]]: + children = [item for item in state.nodes if item.get("parent_id") == node.get("node_id")] + child_count = int(node.get("child_count") or len(children)) + lines: list[tuple[str, int]] = [ + (display_node_name(node), curses.A_BOLD | palette("folder")), + (f"{display_size(node)} across {child_count} immediate entr{'y' if child_count == 1 else 'ies'}.", 0), + ("", 0), + ("WHY THIS MATTERS", curses.A_BOLD | palette("title")), + ] + if children: + largest = max(children, key=node_bytes) + largest_size = node_bytes(largest) + total_size = node_bytes(node) + share = round(100 * largest_size / total_size) if total_size else 0 + lines.append((f"Largest visible area: {display_node_name(largest)} {display_size(largest)} ({share}%).", 0)) + rebuildable = [item for item in children if is_rebuildable_node(item)] + if rebuildable: + candidate = max(rebuildable, key=node_bytes) + lines.append((f"Review candidate: {display_node_name(candidate)} looks rebuildable.", palette("answer"))) + else: + lines.append(("No obvious rebuildable area is loaded yet.", curses.A_DIM | palette("muted"))) + else: + lines.append(("The next level is ready to load on Enter.", curses.A_DIM | palette("muted"))) + if is_rebuildable_node(node): + lines.append(("This selected area itself is marked rebuildable; check its owner first.", palette("answer"))) + lines.extend( + [ + ("", 0), + ("NEXT BEST MOVE", curses.A_BOLD | palette("title")), + ("A ask the local Agent for an explanation.", curses.A_DIM | palette("muted")), + ("Enter open or close this level; dd stages one exact path.", curses.A_DIM | palette("muted")), + ] + ) + return [(wrap_line, attr) for text, attr in lines for wrap_line in wrap_lines(text, max(10, width))] + + def inspector_lines(state: TuiState, width: int) -> list[tuple[str, int]]: node = state.selected() if not node: @@ -1977,28 +2110,13 @@ def inspector_lines(state: TuiState, width: int) -> list[tuple[str, int]]: ] ) else: - lines = [ - *cleanup_lines, - ("ABOUT THIS AREA", curses.A_BOLD | palette("title")), - (display_node_name(node), curses.A_BOLD | palette("folder")), - (f"Path: {display_node_path(node)}", curses.A_DIM), - (f"Space: {display_size(node)}", 0), - (f"Last changed: {display_date(node.get('modified_at'))}", 0), - ("Kind: Folder", 0), - (f"Area: {display_label(node.get('area') or 'unknown scope')}", 0), - (f"Measurement: {display_label(node.get('measurement_status') or 'unknown')}", 0), - ("", 0), - ("WHAT WE KNOW", curses.A_BOLD | palette("title")), - ] - explanation = "We checked this folder's name, size, dates, and visible entries. We did not open file contents, so its exact purpose is not confirmed." - lines.extend([("", 0)]) - for line in wrap_lines(explanation, max(10, width)): - lines.append((line, curses.A_DIM | palette("muted"))) + lines = [*space_story_lines(state, node, width), *cleanup_lines] lines.extend( [ - ("", 0), - ("NEXT STEP", curses.A_BOLD | palette("title")), - ("Press A to ask Codex what this area is likely for.", curses.A_DIM | palette("muted")), + ("PATH", curses.A_BOLD | palette("title")), + (f"{display_node_path(node)}", curses.A_DIM), + (f"Last changed: {display_date(node.get('modified_at'))}", curses.A_DIM), + (f"Measurement: {display_label(node.get('measurement_status') or 'unknown')}", curses.A_DIM), ] ) if state.ai_busy and state.ai_busy_node_id == node_id: @@ -2024,7 +2142,9 @@ def inspector_lines(state: TuiState, width: int) -> list[tuple[str, int]]: def draw_inspector(stdscr: Any, state: TuiState, y: int, height: int, x: int, width: int) -> None: - add_text(stdscr, y, x, "INSPECTOR", width, curses.A_BOLD | palette("title")) + selected = state.selected() + heading = "SPACE STORY" if selected and selected.get("kind") == "folder" else "FILE PREVIEW" + add_text(stdscr, y, x, heading, width, curses.A_BOLD | palette("title")) current_y = y + 1 for text, attr in inspector_lines(state, width)[: max(0, height - 1)]: add_text(stdscr, current_y, x, text, width, attr) diff --git a/src/clean_your_data/cli.py b/src/clean_your_data/cli.py index b954b0c..a8e4407 100644 --- a/src/clean_your_data/cli.py +++ b/src/clean_your_data/cli.py @@ -23,10 +23,25 @@ def normalize_argv(argv: Sequence[str]) -> list[str]: def main(argv: Optional[Sequence[str]] = None) -> int: - """Run the explorer, defaulting to the current directory in the TUI.""" + """Run the explorer or the opt-in local command tracer.""" args = list(sys.argv[1:] if argv is None else argv) if args in (["--version"], ["-V"]): print(f"clean-your-data {__version__}") return 0 + if args and args[0] == "trace": + from .trace import main as trace_main + + return trace_main(args[1:]) + if args and args[0] == "gui": + from .gui import main as gui_main + + return gui_main(args[1:]) + if len(args) >= 2 and args[:2] == ["config", "ai"]: + from .ai_config import config_main + + return config_main(args[2:]) + if args and args[0] == "config": + print("usage: cyd config ai [--show|--auto|--codex|--command COMMAND|--off]", file=sys.stderr) + return 2 return audit_main(normalize_argv(args)) diff --git a/src/clean_your_data/gui.py b/src/clean_your_data/gui.py new file mode 100644 index 0000000..401f5b3 --- /dev/null +++ b/src/clean_your_data/gui.py @@ -0,0 +1,645 @@ +"""Local browser GUI backed by the same scanner and safety gates as the TUI.""" + +from __future__ import annotations + +import argparse +import json +import secrets +import threading +import webbrowser +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from importlib import resources +from pathlib import Path +from typing import Any, Optional, Sequence +from urllib.parse import parse_qs, urlparse + +from .ai_config import ( + config_for_display, + load_ai_config, + parse_command, + save_ai_config, +) +from .audit_local_files import expand_space_map_node, scan_space_map +from .audit_tui import ( + analyze_path_relationships, + ask_local_ai, + build_prompt, + cleanup_gate, + load_cleanup_history, + move_to_trash, + preliminary_cleanup_advice, + read_file_preview, + restore_trash_record, +) + + +GUI_BODY_LIMIT = 32 * 1024 +GUI_QUESTION_LIMIT = 2_000 +GUI_DEFAULT_NODE_LIMIT = 600 +GUI_DEFAULT_DEPTH = 2 +GUI_DEFAULT_TIME_BUDGET = 30 +GUI_DEFAULT_TIMEOUT = 5 +REBUILDABLE_NAMES = { + "node_modules", + ".venv", + "venv", + "build", + "dist", + ".next", + "target", + ".pytest_cache", + "__pycache__", +} + + +class GuiRequestError(Exception): + def __init__(self, message: str, status: int = HTTPStatus.BAD_REQUEST): + super().__init__(message) + self.status = int(status) + + +class GuiSession: + """Mutable local GUI state. Absolute paths never enter public responses.""" + + def __init__( + self, + root: Path, + *, + home: Optional[Path] = None, + depth: int = GUI_DEFAULT_DEPTH, + node_limit: int = GUI_DEFAULT_NODE_LIMIT, + timeout: int = GUI_DEFAULT_TIMEOUT, + time_budget: int = GUI_DEFAULT_TIME_BUDGET, + trash_root: Optional[Path] = None, + history_path: Optional[Path] = None, + ai_config_path: Optional[Path] = None, + ) -> None: + requested = root.expanduser() + try: + resolved = requested.resolve(strict=True) + except (OSError, RuntimeError) as exc: + raise ValueError(f"GUI path is unavailable: {requested}: {exc}") from exc + if not resolved.is_dir(): + raise ValueError(f"GUI path must be a directory: {requested}") + self.root = resolved + self.home = (home or Path.home()).expanduser().resolve() + self.depth = max(0, depth) + self.node_limit = max(10, node_limit) + self.timeout = max(1, timeout) + self.time_budget = max(0, time_budget) + self.trash_root = trash_root + self.history_path = history_path + self.ai_config_path = ai_config_path + self.lock = threading.RLock() + self.nodes: list[dict[str, Any]] = [] + self.nodes_by_id: dict[str, dict[str, Any]] = {} + self.status = "unknown" + self.errors: list[str] = [] + self.staged: set[str] = set() + self.last_records: list[dict[str, Any]] = [] + self.refresh() + + def refresh(self) -> None: + mapped = scan_space_map( + [self.root], + [], + [self.root], + self.home, + self.depth, + self.node_limit, + self.timeout, + self.time_budget, + True, + include_local_paths=True, + allow_skipped_root=True, + ) + with self.lock: + self.nodes = list(mapped.get("nodes") or []) + self.nodes_by_id = {str(node.get("node_id")): node for node in self.nodes} + self.status = str(mapped.get("status") or "unknown") + self.errors = list(mapped.get("errors") or []) + self.staged.intersection_update(self.nodes_by_id) + + def root_node(self) -> dict[str, Any]: + node = next((item for item in self.nodes if item.get("parent_id") is None), None) + if not node: + raise GuiRequestError("The selected path could not be measured.", HTTPStatus.SERVICE_UNAVAILABLE) + return node + + def node(self, node_id: str) -> dict[str, Any]: + with self.lock: + node = self.nodes_by_id.get(str(node_id)) + if not node: + raise GuiRequestError("The selected path is no longer available.", HTTPStatus.NOT_FOUND) + return node + + def public_node(self, node: dict[str, Any]) -> dict[str, Any]: + allowed, reason = self.cleanup_eligibility(node) + name = str(node.get("name") or "") + category = str(node.get("category") or "unknown") + if not allowed: + risk = "protected" + elif category == "cache" or name in REBUILDABLE_NAMES: + risk = "rebuildable" + else: + risk = "review" + return { + key: value + for key, value in node.items() + if not str(key).startswith("_") and key != "measurement_error" + } | { + "risk": risk, + "cleanup_eligible": allowed, + "cleanup_reason": reason, + "staged": str(node.get("node_id")) in self.staged, + } + + def payload(self) -> dict[str, Any]: + with self.lock: + nodes = [self.public_node(node) for node in self.nodes] + staged = list(self.staged) + root = self.public_node(self.root_node()) + provider = config_for_display(load_ai_config(self.ai_config_path)) + return { + "version": 1, + "scope": root, + "status": self.status, + "errors": self.errors[:10], + "nodes": nodes, + "staged": staged, + "ai": provider, + "privacy": { + "redacted_paths": True, + "preview_limit_bytes": 4096, + "preview_sent_to_ai": False, + "server": "127.0.0.1", + }, + } + + def expand(self, node_id: str) -> dict[str, Any]: + node = self.node(node_id) + if node.get("kind") != "folder": + raise GuiRequestError("Only folders can be expanded.") + with self.lock: + loaded = [item for item in self.nodes if item.get("parent_id") == node.get("node_id")] + if loaded: + return {"status": "complete", "nodes": [self.public_node(item) for item in loaded]} + children, status = expand_space_map_node( + node, + self.home, + True, + self.timeout, + self.node_limit, + self.time_budget, + ) + with self.lock: + for child in children: + child_id = str(child.get("node_id")) + if child_id not in self.nodes_by_id: + self.nodes.append(child) + self.nodes_by_id[child_id] = child + return {"status": status, "nodes": [self.public_node(item) for item in children]} + + def preview(self, node_id: str) -> dict[str, Any]: + node = self.node(node_id) + if node.get("kind") == "folder": + with self.lock: + children = [item for item in self.nodes if item.get("parent_id") == node.get("node_id")] + if children: + lines = [ + f"{item.get('name')}{'/' if item.get('kind') == 'folder' else ''} {item.get('human_size') or 'unknown'}" + for item in children[:24] + ] + if len(children) > 24: + lines.append(f"... {len(children) - 24} more loaded entries ...") + elif node.get("can_expand"): + lines = ["Open this folder to load its next level."] + else: + lines = ["(empty folder)"] + return {"kind": "folder", "lines": lines, "limited": len(children) > 24} + lines = read_file_preview(node) + return {"kind": "file", "lines": lines, "limited": any("preview limited" in line for line in lines)} + + def inspect(self, node_id: str) -> dict[str, Any]: + node = self.node(node_id) + allowed, reason = self.cleanup_eligibility(node) + return { + "node": self.public_node(node), + "advice": preliminary_cleanup_advice(node), + "cleanup": {"eligible": allowed, "reason": reason}, + "ai_context": { + "includes": ["redacted path", "name", "kind", "size", "modified time", "category", "measurement status"], + "excludes": ["file preview", "file contents", "credentials", "cleanup authority"], + }, + } + + def relationships(self, node_id: str) -> dict[str, Any]: + return analyze_path_relationships(self.node(node_id)) + + def cleanup_eligibility(self, node: dict[str, Any]) -> tuple[bool, str]: + if str(node.get("node_id")) == str(self.root_node().get("node_id")): + return False, "the active GUI scope is protected" + local_path = Path(str(node.get("_local_path") or "")) + try: + local_path.resolve().relative_to(self.root) + except (OSError, RuntimeError, ValueError): + return False, "the path is outside the active GUI scope" + return cleanup_gate(node) + + def toggle_stage(self, node_id: str) -> dict[str, Any]: + node = self.node(node_id) + allowed, reason = self.cleanup_eligibility(node) + if not allowed: + raise GuiRequestError(f"This path cannot enter the cleanup basket: {reason}", HTTPStatus.CONFLICT) + with self.lock: + if node_id in self.staged: + self.staged.remove(node_id) + staged = False + else: + self.staged.add(node_id) + staged = True + return {"node_id": node_id, "staged": staged, "basket": self.basket()} + + def basket(self) -> dict[str, Any]: + with self.lock: + selected = [self.nodes_by_id[node_id] for node_id in self.staged if node_id in self.nodes_by_id] + return { + "nodes": [self.public_node(node) for node in selected], + "total_bytes": sum(int(node.get("allocated_bytes") or 0) for node in selected), + } + + def ask(self, node_id: str, question: str) -> dict[str, Any]: + node = self.node(node_id) + clean_question = question.strip() + if not clean_question: + raise GuiRequestError("Question cannot be empty.") + if len(clean_question) > GUI_QUESTION_LIMIT: + raise GuiRequestError(f"Question must be at most {GUI_QUESTION_LIMIT} characters.") + answer, message = ask_local_ai( + build_prompt(node, clean_question), + config_path=self.ai_config_path, + ) + if answer is None: + provider = config_for_display(load_ai_config(self.ai_config_path)).get("provider") or "Configured Agent" + raise GuiRequestError( + f"{provider} did not return an answer. Check the local provider configuration and try again.", + HTTPStatus.SERVICE_UNAVAILABLE, + ) + return {"node_id": node_id, "answer": answer, "provider_message": message} + + def move_staged_to_trash(self, confirmations: dict[str, str]) -> dict[str, Any]: + with self.lock: + staged_ids = sorted( + self.staged, + key=lambda node_id: str(self.nodes_by_id.get(node_id, {}).get("path") or node_id), + ) + if not staged_ids: + raise GuiRequestError("The cleanup basket is empty.", HTTPStatus.CONFLICT) + + staged_nodes = [self.node(node_id) for node_id in staged_ids] + for node in staged_nodes: + node_id = str(node.get("node_id")) + if confirmations.get(node_id) != str(node.get("path")): + raise GuiRequestError("Exact path confirmation did not match.", HTTPStatus.CONFLICT) + allowed, reason = self.cleanup_eligibility(node) + if not allowed: + raise GuiRequestError(f"{node.get('path')}: {reason}", HTTPStatus.CONFLICT) + + records: list[dict[str, Any]] = [] + errors: list[dict[str, str]] = [] + for node in staged_nodes: + try: + record = move_to_trash( + Path(str(node.get("_local_path"))), + node=node, + trash_root=self.trash_root, + history_path=self.history_path, + ) + except (OSError, RuntimeError, ValueError): + errors.append( + { + "path": str(node.get("path") or "unknown"), + "error": "The path could not be moved. It may have changed or become unavailable.", + } + ) + continue + records.append(record) + with self.lock: + self.staged.discard(str(node.get("node_id"))) + self.last_records = records + self.refresh() + return { + "moved": len(records), + "errors": errors, + "records": [ + { + "record_id": record.get("record_id"), + "name": record.get("name"), + "human_size": record.get("human_size"), + "status": record.get("status"), + } + for record in records + ], + "session": self.payload(), + } + + def undo(self) -> dict[str, Any]: + if not self.last_records: + raise GuiRequestError("There is no cleanup action to undo.", HTTPStatus.CONFLICT) + pending_records = list(self.last_records) + restored: list[dict[str, Any]] = [] + errors: list[dict[str, str]] = [] + failed_record_ids: set[str] = set() + for record in reversed(pending_records): + try: + restored.append(restore_trash_record(record, self.history_path)) + except (OSError, RuntimeError, ValueError): + record_id = str(record.get("record_id") or "") + failed_record_ids.add(record_id) + errors.append( + { + "name": str(record.get("name") or "unknown"), + "error": "The Trash item could not be restored. Its original path may now be occupied.", + } + ) + self.last_records = [ + record for record in pending_records if str(record.get("record_id") or "") in failed_record_ids + ] + self.refresh() + return { + "restored": len(restored), + "errors": errors, + "records": [{"record_id": row.get("record_id"), "name": row.get("name")} for row in restored], + "session": self.payload(), + } + + def cleanup_history(self) -> list[dict[str, Any]]: + rows = load_cleanup_history(self.history_path) + return [ + { + "record_id": row.get("record_id"), + "name": row.get("name"), + "kind": row.get("kind"), + "human_size": row.get("human_size"), + "moved_at": row.get("moved_at"), + "restored_at": row.get("restored_at"), + "status": row.get("status"), + } + for row in rows[-50:] + ] + + def set_ai_config(self, mode: str, command_text: str = "") -> dict[str, Any]: + normalized_mode = mode.strip().lower() + if normalized_mode not in {"auto", "codex", "command", "off"}: + raise GuiRequestError("Unknown AI mode.") + try: + command = parse_command(command_text) if normalized_mode == "command" else [] + except ValueError as exc: + raise GuiRequestError(str(exc)) from exc + config = save_ai_config( + {"version": 1, "mode": normalized_mode, "command": command}, + self.ai_config_path, + ) + return config_for_display(config) + + +def load_gui_html(token: str) -> bytes: + template = resources.files("clean_your_data").joinpath("web/index.html").read_text(encoding="utf-8") + return template.replace("__CYD_SESSION_TOKEN__", token).encode("utf-8") + + +class GuiHTTPServer(ThreadingHTTPServer): + daemon_threads = True + allow_reuse_address = True + + def __init__( + self, + server_address: tuple[str, int], + session: GuiSession, + token: str, + *, + verbose: bool = False, + ) -> None: + self.session = session + self.token = token + self.verbose = verbose + self.html = load_gui_html(token) + super().__init__(server_address, GuiRequestHandler) + + +class GuiRequestHandler(BaseHTTPRequestHandler): + server: GuiHTTPServer + + def log_message(self, format: str, *args: Any) -> None: + if self.server.verbose: + super().log_message(format, *args) + + def _headers(self, status: int, content_type: str, length: int) -> None: + self.send_response(status) + self.send_header("Content-Type", content_type) + self.send_header("Content-Length", str(length)) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header("X-Frame-Options", "DENY") + self.send_header("Referrer-Policy", "no-referrer") + self.send_header( + "Content-Security-Policy", + "default-src 'self'; script-src 'unsafe-inline'; style-src 'unsafe-inline'; " + "connect-src 'self'; img-src 'none'; object-src 'none'; base-uri 'none'; " + "frame-ancestors 'none'; form-action 'self'", + ) + self.end_headers() + + def _send_json(self, payload: Any, status: int = HTTPStatus.OK) -> None: + body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + self._headers(status, "application/json; charset=utf-8", len(body)) + self.wfile.write(body) + + def _send_error_json(self, exc: Exception) -> None: + if isinstance(exc, GuiRequestError): + status = exc.status + message = str(exc) + else: + status = HTTPStatus.INTERNAL_SERVER_ERROR + message = "The local GUI could not complete this request." + self._send_json({"error": message}, status) + + def _authorized(self) -> bool: + return secrets.compare_digest(self.headers.get("X-CYD-Token", ""), self.server.token) + + def _check_local_origin(self) -> bool: + origin = self.headers.get("Origin") + if not origin: + return True + parsed = urlparse(origin) + return parsed.scheme == "http" and parsed.hostname in {"127.0.0.1", "localhost"} and parsed.port == self.server.server_port + + def _check_local_host(self) -> bool: + host = self.headers.get("Host", "") + try: + parsed = urlparse("//" + host) + port = parsed.port + except ValueError: + return False + return parsed.hostname in {"127.0.0.1", "localhost"} and ( + port == self.server.server_port or (port is None and self.server.server_port == 80) + ) + + def _json_body(self) -> dict[str, Any]: + content_type = self.headers.get("Content-Type", "") + if not content_type.startswith("application/json"): + raise GuiRequestError("Requests must use application/json.", HTTPStatus.UNSUPPORTED_MEDIA_TYPE) + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError as exc: + raise GuiRequestError("Invalid request length.") from exc + if length < 0 or length > GUI_BODY_LIMIT: + raise GuiRequestError("Request body is too large.", HTTPStatus.REQUEST_ENTITY_TOO_LARGE) + try: + payload = json.loads(self.rfile.read(length).decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise GuiRequestError("Request body is not valid JSON.") from exc + if not isinstance(payload, dict): + raise GuiRequestError("Request body must be a JSON object.") + return payload + + def _require_api_access(self) -> bool: + if not self._check_local_host() or not self._authorized() or not self._check_local_origin(): + self._send_json({"error": "Local GUI session authorization failed."}, HTTPStatus.FORBIDDEN) + return False + return True + + def do_GET(self) -> None: + parsed = urlparse(self.path) + if parsed.path == "/": + if not self._check_local_host(): + self._send_json({"error": "Local GUI host validation failed."}, HTTPStatus.FORBIDDEN) + return + body = self.server.html + self._headers(HTTPStatus.OK, "text/html; charset=utf-8", len(body)) + self.wfile.write(body) + return + if not self._require_api_access(): + return + try: + query = parse_qs(parsed.query) + if parsed.path == "/api/session": + self._send_json(self.server.session.payload()) + elif parsed.path == "/api/preview": + self._send_json(self.server.session.preview(_one(query, "node_id"))) + elif parsed.path == "/api/inspect": + self._send_json(self.server.session.inspect(_one(query, "node_id"))) + elif parsed.path == "/api/relationships": + self._send_json(self.server.session.relationships(_one(query, "node_id"))) + elif parsed.path == "/api/config": + self._send_json(config_for_display(load_ai_config(self.server.session.ai_config_path))) + elif parsed.path == "/api/history": + self._send_json({"history": self.server.session.cleanup_history()}) + else: + self._send_json({"error": "Not found."}, HTTPStatus.NOT_FOUND) + except Exception as exc: + self._send_error_json(exc) + + def do_POST(self) -> None: + parsed = urlparse(self.path) + if not self._require_api_access(): + return + try: + payload = self._json_body() + if parsed.path == "/api/expand": + result = self.server.session.expand(str(payload.get("node_id") or "")) + elif parsed.path == "/api/stage": + result = self.server.session.toggle_stage(str(payload.get("node_id") or "")) + elif parsed.path == "/api/ask": + result = self.server.session.ask( + str(payload.get("node_id") or ""), + str(payload.get("question") or ""), + ) + elif parsed.path == "/api/config": + result = self.server.session.set_ai_config( + str(payload.get("mode") or ""), + str(payload.get("command") or ""), + ) + elif parsed.path == "/api/trash": + confirmations = payload.get("confirmations") + if not isinstance(confirmations, dict): + raise GuiRequestError("Exact path confirmations are required.") + result = self.server.session.move_staged_to_trash( + {str(key): str(value) for key, value in confirmations.items()} + ) + elif parsed.path == "/api/undo": + result = self.server.session.undo() + elif parsed.path == "/api/rescan": + self.server.session.refresh() + result = self.server.session.payload() + elif parsed.path == "/api/shutdown": + result = {"stopping": True} + threading.Thread(target=self.server.shutdown, daemon=True).start() + else: + self._send_json({"error": "Not found."}, HTTPStatus.NOT_FOUND) + return + self._send_json(result) + except Exception as exc: + self._send_error_json(exc) + + +def _one(query: dict[str, list[str]], name: str) -> str: + values = query.get(name) or [] + if len(values) != 1 or not values[0]: + raise GuiRequestError(f"Missing query parameter: {name}") + return values[0] + + +def create_server( + session: GuiSession, + *, + port: int = 0, + token: Optional[str] = None, + verbose: bool = False, +) -> GuiHTTPServer: + if port < 0 or port > 65535: + raise ValueError("port must be between 0 and 65535") + return GuiHTTPServer(("127.0.0.1", port), session, token or secrets.token_urlsafe(32), verbose=verbose) + + +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="cyd gui", + description="Open the local Clean Your Data browser GUI.", + ) + parser.add_argument("path", nargs="?", default=".", help="Directory to explore. Defaults to the current directory.") + parser.add_argument("--no-open", action="store_true", help="Start the local GUI server without opening a browser.") + parser.add_argument("--port", type=int, default=0, help="Loopback port. Defaults to a random available port.") + parser.add_argument("--depth", type=int, default=GUI_DEFAULT_DEPTH, help="Initial scan depth. Deeper folders load on demand.") + parser.add_argument("--node-limit", type=int, default=GUI_DEFAULT_NODE_LIMIT, help="Maximum nodes in the initial map.") + parser.add_argument("--time-budget", type=int, default=GUI_DEFAULT_TIME_BUDGET, help="Initial scan time budget in seconds.") + parser.add_argument("--verbose", action="store_true", help="Print local HTTP request logs.") + return parser.parse_args(list(argv)) + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = parse_args(list(argv or [])) + try: + session = GuiSession( + Path(args.path), + depth=args.depth, + node_limit=args.node_limit, + time_budget=args.time_budget, + ) + server = create_server(session, port=args.port, verbose=args.verbose) + except (OSError, ValueError) as exc: + print(f"error: {exc}") + return 2 + url = f"http://127.0.0.1:{server.server_port}/" + print(f"Clean Your Data GUI: {url}") + print("Local only. Press Ctrl-C to stop the server.") + if not args.no_open: + webbrowser.open(url) + try: + server.serve_forever(poll_interval=0.25) + except KeyboardInterrupt: + print("\nStopping Clean Your Data GUI.") + finally: + server.server_close() + return 0 diff --git a/src/clean_your_data/trace.py b/src/clean_your_data/trace.py new file mode 100644 index 0000000..3bba795 --- /dev/null +++ b/src/clean_your_data/trace.py @@ -0,0 +1,570 @@ +"""Trace a local command and report metadata-only file changes. + +This is a bounded, opt-in observer. It does not read file contents, inspect +environment variables, or claim kernel-level process attribution. Changes are +associated with the traced command because they were observed in its scope +while that command was running. +""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import signal +import sqlite3 +import stat as stat_module +import subprocess +import sys +import time +import uuid +from dataclasses import dataclass +from datetime import datetime +from pathlib import Path +from typing import Any, Optional, Sequence + + +TRACE_SCHEMA_VERSION = "1.0" +DEFAULT_INTERVAL_SECONDS = 0.5 +DEFAULT_MAX_ENTRIES = 20_000 +TRACE_STATE_ENV = "CLEAN_YOUR_DATA_STATE_DIR" +TRACE_DB_NAME = "provenance.sqlite3" + + +@dataclass(frozen=True) +class EntryState: + """The small set of stat fields needed to detect a local change.""" + + kind: str + size: Optional[int] + mtime_ns: Optional[int] + mode: int + inode: int + + +@dataclass +class Snapshot: + entries: dict[str, EntryState] + roots: list[Path] + scanned_entries: int + skipped_paths: list[str] + limited: bool + + +def timestamp() -> str: + return datetime.now().astimezone().isoformat(timespec="seconds") + + +def session_id() -> str: + return "trace-" + time.strftime("%Y%m%d-%H%M%S") + "-" + uuid.uuid4().hex[:8] + + +def entry_state(stat_result: os.stat_result) -> EntryState: + mode = stat_result.st_mode + if stat_module.S_ISDIR(mode): + kind = "folder" + size: Optional[int] = None + elif stat_module.S_ISREG(mode): + kind = "file" + size = int(stat_result.st_size) + elif stat_module.S_ISLNK(mode): + kind = "symlink" + size = int(stat_result.st_size) + else: + kind = "other" + size = int(stat_result.st_size) + return EntryState( + kind=kind, + size=size, + mtime_ns=int(getattr(stat_result, "st_mtime_ns", int(stat_result.st_mtime * 1_000_000_000))), + mode=int(mode & 0o7777), + inode=int(getattr(stat_result, "st_ino", 0)), + ) + + +def snapshot(roots: Sequence[Path], max_entries: int) -> Snapshot: + """Walk roots without following symlinks or reading file contents.""" + entries: dict[str, EntryState] = {} + skipped: list[str] = [] + limited = False + pending: list[Path] = list(reversed(list(roots))) + + while pending: + path = pending.pop() + key = str(path) + if key in entries: + continue + if len(entries) >= max_entries: + limited = True + break + try: + stat_result = path.lstat() + except OSError as exc: + if len(skipped) < 30: + skipped.append(f"{path}: {exc}") + continue + state = entry_state(stat_result) + entries[key] = state + if state.kind != "folder": + continue + try: + children = sorted(path.iterdir(), key=lambda item: item.name.casefold()) + except OSError as exc: + if len(skipped) < 30: + skipped.append(f"{path}: {exc}") + continue + for child in reversed(children): + if len(entries) + len(pending) >= max_entries: + limited = True + break + pending.append(child) + + return Snapshot( + entries=entries, + roots=list(roots), + scanned_entries=len(entries), + skipped_paths=skipped, + limited=limited, + ) + + +def normalize_roots(raw_roots: Sequence[str], cwd: Path) -> list[Path]: + requested = [Path(value).expanduser() for value in raw_roots] or [cwd] + roots: list[Path] = [] + for path in requested: + try: + resolved = path.resolve() + except RuntimeError as exc: + raise ValueError(f"trace path could not be resolved: {path}") from exc + if not resolved.exists(): + raise ValueError(f"trace path does not exist: {path}") + if resolved not in roots: + roots.append(resolved) + return roots + + +def state_dict(state: Optional[EntryState]) -> Optional[dict[str, Any]]: + if state is None: + return None + return { + "kind": state.kind, + "size": state.size, + "mtime_ns": state.mtime_ns, + "mode": oct(state.mode), + "inode": state.inode, + } + + +def changed_fields(before: EntryState, after: EntryState) -> list[str]: + fields: list[str] = [] + if before.kind != after.kind: + fields.append("kind") + if before.size != after.size: + fields.append("size") + if before.mtime_ns != after.mtime_ns: + fields.append("modified time") + if before.mode != after.mode: + fields.append("permissions") + if before.inode != after.inode: + fields.append("inode") + return fields + + +def record_event( + events: dict[str, dict[str, Any]], + path: str, + event_type: str, + before: Optional[EntryState], + after: Optional[EntryState], + observed_at: str, + fields: Optional[list[str]] = None, +) -> None: + event = events.get(path) + if event is None: + event = { + "absolute_path": path, + "kind": (after or before).kind if (after or before) else "unknown", + "event_types": [], + "changed_fields": [], + "first_observed_at": observed_at, + "last_observed_at": observed_at, + "before": state_dict(before), + "after": state_dict(after), + } + events[path] = event + if event_type not in event["event_types"]: + event["event_types"].append(event_type) + if fields: + for field in fields: + if field not in event["changed_fields"]: + event["changed_fields"].append(field) + event["last_observed_at"] = observed_at + event["after"] = state_dict(after) + if after is not None: + event["kind"] = after.kind + + +def observe_delta( + previous: Snapshot, + current: Snapshot, + baseline: Snapshot, + events: dict[str, dict[str, Any]], +) -> None: + observed_at = timestamp() + paths = sorted(set(previous.entries) | set(current.entries)) + for path in paths: + old = previous.entries.get(path) + new = current.entries.get(path) + before = baseline.entries.get(path) + if old is None and new is not None: + record_event(events, path, "created", before, new, observed_at) + continue + if old is not None and new is None: + record_event(events, path, "deleted", before, None, observed_at) + continue + if old is None or new is None: + continue + fields = changed_fields(old, new) + # Directory mtimes change when a child changes. Report the child event, + # not every ancestor as a misleading file modification. + if fields and not (old.kind == "folder" and new.kind == "folder"): + record_event(events, path, "modified", before, new, observed_at, fields) + + +def display_path(path: str, roots: Sequence[Path], redact: bool) -> str: + candidate = Path(path) + for root in roots: + try: + relative = candidate.relative_to(root) + return "." if not relative.parts else "./" + str(relative) + except ValueError: + continue + if redact: + home = Path.home().resolve() + try: + return "~/" + str(candidate.relative_to(home)) + except ValueError: + return "" + return str(candidate) + + +def public_event(event: dict[str, Any], roots: Sequence[Path], redact: bool) -> dict[str, Any]: + result = dict(event) + absolute = result.pop("absolute_path") + result["path"] = display_path(absolute, roots, redact) + return result + + +def public_report(report: dict[str, Any], roots: Sequence[Path], redact: bool) -> dict[str, Any]: + result = dict(report) + session = dict(result["session"]) + if redact: + home_text = str(Path.home().resolve()) + session["command"] = [str(item).replace(home_text, "~") for item in session["command"]] + session["cwd"] = display_path(str(Path(session["cwd"]).resolve()), roots, redact) + session["scope_roots"] = [display_path(str(root), roots, redact) for root in roots] + result["session"] = session + observation = dict(result["observation"]) + safe_skipped: list[str] = [] + for item in observation.get("skipped_paths") or []: + path_text, separator, detail = str(item).partition(": ") + shown_path = display_path(path_text, roots, redact) + safe_skipped.append(f"{shown_path}{separator}{detail}" if separator else shown_path) + observation["skipped_paths"] = safe_skipped + result["observation"] = observation + result["events"] = [public_event(event, roots, redact) for event in report["events"]] + return result + + +def persist_trace( + report: dict[str, Any], + store_path: Path, +) -> Optional[str]: + """Persist local provenance records without sending them anywhere.""" + try: + store_path.parent.mkdir(parents=True, exist_ok=True) + try: + os.chmod(store_path.parent, 0o700) + except OSError: + pass + descriptor = os.open(store_path, os.O_CREAT | os.O_APPEND, 0o600) + os.close(descriptor) + try: + os.chmod(store_path, 0o600) + except OSError: + pass + with sqlite3.connect(store_path) as connection: + connection.execute( + """ + CREATE TABLE IF NOT EXISTS trace_sessions ( + session_id TEXT PRIMARY KEY, + started_at TEXT NOT NULL, + finished_at TEXT NOT NULL, + command_json TEXT NOT NULL, + cwd TEXT NOT NULL, + roots_json TEXT NOT NULL, + return_code INTEGER, + observed_count INTEGER NOT NULL + ) + """ + ) + connection.execute( + """ + CREATE TABLE IF NOT EXISTS trace_events ( + event_id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + path TEXT NOT NULL, + kind TEXT NOT NULL, + event_types_json TEXT NOT NULL, + changed_fields_json TEXT NOT NULL, + first_observed_at TEXT NOT NULL, + last_observed_at TEXT NOT NULL, + before_json TEXT, + after_json TEXT, + FOREIGN KEY(session_id) REFERENCES trace_sessions(session_id) + ) + """ + ) + session = report["session"] + connection.execute( + """ + INSERT OR REPLACE INTO trace_sessions + (session_id, started_at, finished_at, command_json, cwd, roots_json, return_code, observed_count) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + session["id"], + session["started_at"], + session["finished_at"], + json.dumps(session["command"], ensure_ascii=False), + session["cwd"], + json.dumps(session["scope_roots"], ensure_ascii=False), + session["return_code"], + len(report["events"]), + ), + ) + connection.execute("DELETE FROM trace_events WHERE session_id = ?", (session["id"],)) + for event in report["events"]: + connection.execute( + """ + INSERT INTO trace_events + (session_id, path, kind, event_types_json, changed_fields_json, + first_observed_at, last_observed_at, before_json, after_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + session["id"], + event["absolute_path"], + event["kind"], + json.dumps(event["event_types"], ensure_ascii=False), + json.dumps(event["changed_fields"], ensure_ascii=False), + event["first_observed_at"], + event["last_observed_at"], + json.dumps(event["before"], ensure_ascii=False) if event["before"] else None, + json.dumps(event["after"], ensure_ascii=False) if event["after"] else None, + ), + ) + except (OSError, sqlite3.Error) as exc: + return str(exc) + return None + + +def run_trace( + command: Sequence[str], + roots: Sequence[Path], + cwd: Path, + interval: float, + max_entries: int, +) -> tuple[dict[str, Any], dict[str, Any]]: + started_at = timestamp() + started = time.monotonic() + trace_id = session_id() + before = snapshot(roots, max_entries) + events: dict[str, dict[str, Any]] = {} + process: Optional[subprocess.Popen[bytes]] = None + interrupted = False + return_code: Optional[int] = None + try: + process = subprocess.Popen( + list(command), + cwd=str(cwd), + start_new_session=(os.name != "nt"), + ) + previous = before + while True: + return_code = process.poll() + if return_code is not None: + break + time.sleep(interval) + current = snapshot(roots, max_entries) + observe_delta(previous, current, before, events) + previous = current + after = snapshot(roots, max_entries) + observe_delta(previous, after, before, events) + except KeyboardInterrupt: + interrupted = True + if process is not None and process.poll() is None: + try: + if os.name != "nt": + os.killpg(process.pid, signal.SIGTERM) + else: + process.terminate() + except OSError: + pass + process.wait() + return_code = 130 + after = snapshot(roots, max_entries) + observe_delta(before, after, before, events) + except OSError: + raise + + finished_at = timestamp() + report = { + "trace_schema_version": TRACE_SCHEMA_VERSION, + "session": { + "id": trace_id, + "command": list(command), + "cwd": str(cwd), + "scope_roots": [str(root) for root in roots], + "pid": process.pid if process is not None else None, + "started_at": started_at, + "finished_at": finished_at, + "duration_seconds": round(time.monotonic() - started, 3), + "return_code": return_code, + "interrupted": interrupted, + }, + "observation": { + "method": "metadata snapshots while the traced command runs", + "attribution": "associated with the traced command; not kernel-level proof", + "interval_seconds": interval, + "max_entries": max_entries, + "before_entries": before.scanned_entries, + "after_entries": after.scanned_entries, + "limited": before.limited or after.limited, + "skipped_paths": before.skipped_paths[:15] + after.skipped_paths[:15], + }, + "events": sorted(events.values(), key=lambda item: item["absolute_path"]), + } + return report, {"before": before, "after": after} + + +def render_text(report: dict[str, Any], roots: Sequence[Path], redact: bool, store_path: Optional[Path], store_error: Optional[str]) -> str: + public = public_report(report, roots, redact) + session = public["session"] + observation = public["observation"] + lines = [ + "CLEAN YOUR DATA / TRACE", + f"Session: {session['id']}", + f"Command: {shlex.join(session['command'])}", + f"Scope: {', '.join(session['scope_roots'])}", + f"Result: exit {session['return_code']} | {session['duration_seconds']}s | {len(public['events'])} changed path(s)", + "", + "ATTRIBUTION", + "Changes below were observed inside the selected scope while this command was running.", + "This is evidence of association, not kernel-level proof of the exact writer process.", + "", + ] + if not public["events"]: + lines.append("No changed paths were observed.") + else: + for event in public["events"]: + types = "+".join(event["event_types"]).upper() + details = f" [{', '.join(event['changed_fields'])}]" if event["changed_fields"] else "" + lines.append(f"{types:<18} {event['path']} ({event['kind']}){details}") + lines.extend( + [ + "", + "OBSERVATION", + f"Metadata entries: {observation['before_entries']} before, {observation['after_entries']} after", + f"Polling interval: {observation['interval_seconds']}s", + ] + ) + if observation["limited"]: + lines.append("Warning: the snapshot reached its entry limit; this trace is partial.") + if observation["skipped_paths"]: + lines.append(f"Warning: {len(observation['skipped_paths'])} paths could not be read.") + if store_path and not store_error: + lines.append(f"Saved locally: {display_path(str(store_path), roots, redact)}") + elif store_error: + lines.append(f"Could not save the local trace record: {store_error}") + return "\n".join(lines) + + +def state_dir() -> Path: + configured = os.environ.get(TRACE_STATE_ENV, "").strip() + return Path(configured).expanduser() if configured else Path.home() / ".clean-your-data" + + +def parse_args(argv: Sequence[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="cyd trace", + description="Run a local command and record metadata-only file changes in selected paths.", + ) + parser.add_argument("--path", action="append", default=[], help="Directory or file to observe. May be repeated. Defaults to the command working directory.") + parser.add_argument("--cwd", help="Working directory for the traced command. Defaults to the first --path or the current directory.") + parser.add_argument("--interval", type=float, default=DEFAULT_INTERVAL_SECONDS, help="Seconds between metadata snapshots. Defaults to 0.5.") + parser.add_argument("--max-entries", type=int, default=DEFAULT_MAX_ENTRIES, help="Maximum entries per snapshot. Defaults to 20000.") + parser.add_argument("--format", choices=["text", "json"], default="text", help="Trace report format.") + parser.add_argument("--output", help="Write the report to this path instead of stdout.") + parser.add_argument("--state-dir", help="Local directory for the provenance SQLite record.") + parser.add_argument("--no-redact", action="store_true", help="Show absolute paths outside the selected scope.") + parser.add_argument("command", nargs=argparse.REMAINDER, help="Command to run after `--`, for example `codex` or `claude`.") + args = parser.parse_args(list(argv)) + command = list(args.command) + if command and command[0] == "--": + command = command[1:] + if not command: + parser.error("a command is required after `--`, for example: cyd trace -- codex") + if args.interval <= 0: + parser.error("--interval must be greater than zero") + if args.max_entries <= 0: + parser.error("--max-entries must be greater than zero") + args.command = command + return args + + +def main(argv: Optional[Sequence[str]] = None) -> int: + args = parse_args(sys.argv[1:] if argv is None else argv) + try: + requested_cwd = Path(args.cwd).expanduser().resolve() if args.cwd else None + default_cwd = requested_cwd or Path.cwd().resolve() + roots = normalize_roots(args.path, default_cwd) + cwd = requested_cwd or (roots[0] if roots[0].is_dir() else roots[0].parent) + if not cwd.is_dir(): + raise ValueError(f"trace working directory is not a directory: {cwd}") + report, _ = run_trace(args.command, roots, cwd, args.interval, args.max_entries) + except (OSError, ValueError) as exc: + print(f"trace error: {exc}", file=sys.stderr) + return 2 + + store_root = Path(args.state_dir).expanduser() if args.state_dir else state_dir() + store_path = store_root / TRACE_DB_NAME + store_error = persist_trace(report, store_path) + public = public_report(report, roots, not args.no_redact) + public["store"] = { + "saved": store_error is None, + "path": display_path(str(store_path), roots, not args.no_redact), + } + if store_error: + public["store"]["error"] = store_error if args.no_redact else "The local trace record could not be saved." + + if args.format == "json": + output = json.dumps(public, ensure_ascii=False, indent=2) + else: + output = render_text(report, roots, not args.no_redact, store_path, store_error) + if args.output: + output_path = Path(args.output).expanduser() + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(output + "\n", encoding="utf-8") + try: + os.chmod(output_path, 0o600) + except OSError: + pass + else: + print(output) + return int(report["session"]["return_code"] or 0) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/clean_your_data/web/index.html b/src/clean_your_data/web/index.html new file mode 100644 index 0000000..cb777eb --- /dev/null +++ b/src/clean_your_data/web/index.html @@ -0,0 +1,1353 @@ + + + + + + Clean Your Data + + + +
+ + +
+
+
+ + +
Loading local scope…
+
+
+ + + + +
+
+ +
+
+
+
Loading…Preparing metadata map
+
+ + + +
+
+
+
Measured—
+
Loaded map—
+
Cleanup basket0 B
+
+
+
Name
Relative space
Size
Modified
+
Scanning the selected path…
+
+
+ + +
+ +
+
+
⌫
+
Cleanup basket is emptyInspect an exact path before staging it.
+
+
+ + +
+
+
+
+ +
+ + + + + + + + + + + + diff --git a/tests/cleanup_test.py b/tests/cleanup_test.py index 0920f96..1d9052b 100644 --- a/tests/cleanup_test.py +++ b/tests/cleanup_test.py @@ -61,8 +61,75 @@ def rescan(): ) assert tui.cleanup_gate(node)[0] is True assert tui.cleanup_gate(dict(node, category="app-state"))[0] is False + protected_root = root / "project" + protected_root.mkdir() + root_node = dict( + node, + node_id="root-node", + name="project", + kind="folder", + _local_path=str(protected_root), + ) + assert tui.cleanup_gate(root_node)[0] is False + assert "scope" in tui.cleanup_gate(root_node)[1] + + git_dir = protected_root / ".git" + git_dir.mkdir() + git_node = dict( + root_node, + node_id="git-node", + parent_id="root-node", + name=".git", + _local_path=str(git_dir), + ) + assert tui.cleanup_gate(git_node)[0] is False + assert "protected" in tui.cleanup_gate(git_node)[1] + + env_file = protected_root / ".env" + env_file.write_text("TOKEN=fixture\n", encoding="utf-8") + env_node = dict( + node, + node_id="env-node", + parent_id="root-node", + name=".env", + _local_path=str(env_file), + ) + assert tui.cleanup_gate(env_node)[0] is False + assert "credential" in tui.cleanup_gate(env_node)[1] + production_env = protected_root / ".env.production" + production_env.write_text("TOKEN=fixture\n", encoding="utf-8") + production_env_node = dict(env_node, name=production_env.name, _local_path=str(production_env)) + assert tui.cleanup_gate(production_env_node)[0] is False assert "preliminary" in tui.cleanup_prompt(node).lower() + original = root / "replace-me.txt" + original.write_text("original\n", encoding="utf-8") + original_stat = original.lstat() + replacement_node = dict( + node, + node_id="replacement-node", + name=original.name, + _local_path=str(original), + _stat_device=original_stat.st_dev, + _stat_inode=original_stat.st_ino, + _stat_mode=original_stat.st_mode, + _stat_ctime_ns=getattr( + original_stat, + "st_ctime_ns", + int(original_stat.st_ctime * 1_000_000_000), + ), + ) + original.unlink() + original.write_text("replacement\n", encoding="utf-8") + assert tui.cleanup_gate(replacement_node)[0] is False + assert "changed since" in tui.cleanup_gate(replacement_node)[1] + try: + tui.move_to_trash(Path(replacement_node["_local_path"]), node=replacement_node, trash_root=trash, history_path=history) + except ValueError as exc: + assert "changed since" in str(exc) + else: + raise AssertionError("the Trash operation must re-check path identity immediately before moving") + assert tui.handle_key(None, state, "d", 80) is True assert state.vim_pending_d is True assert tui.handle_key(None, state, "d", 80) is True @@ -90,6 +157,33 @@ def rescan(): assert source.exists() assert [item["name"] for item in state.nodes] == [source.name] assert state.cleanup_history[-1]["status"] == "restored" + + first_record = tui.move_to_trash(source, node=node, trash_root=trash, history_path=history) + tui.restore_trash_record(first_record, history) + second_record = tui.move_to_trash(source, node=node, trash_root=trash, history_path=history) + assert first_record["record_id"] != second_record["record_id"] + tui.restore_trash_record(second_record, history) + + rollback_record = tui.move_to_trash(source, node=node, trash_root=trash, history_path=history) + function_globals = tui.restore_trash_record.__globals__ + original_save_history = function_globals["save_cleanup_history"] + + def fail_restore_history(*_args, **_kwargs): + raise OSError("fixture history failure") + + function_globals["save_cleanup_history"] = fail_restore_history + try: + try: + tui.restore_trash_record(rollback_record, history) + except RuntimeError as exc: + assert "rolled back" in str(exc) + else: + raise AssertionError("a failed history write must roll the restore back into Trash") + finally: + function_globals["save_cleanup_history"] = original_save_history + assert not source.exists() + assert Path(str(rollback_record["trash_path"])).exists() + tui.restore_trash_record(rollback_record, history) finally: if previous_ai is None: os.environ.pop("CLEAN_YOUR_DATA_AI_COMMAND", None) diff --git a/tests/gui_test.py b/tests/gui_test.py new file mode 100644 index 0000000..8e7db49 --- /dev/null +++ b/tests/gui_test.py @@ -0,0 +1,298 @@ +#!/usr/bin/env python3 +"""Integration tests for the local browser GUI and its safety boundary.""" + +from __future__ import annotations + +import json +import os +import tempfile +import threading +import urllib.error +import urllib.request +from pathlib import Path +from typing import Optional + + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" + + +def request_json( + url: str, + token: str, + *, + method: str = "GET", + payload: Optional[dict[str, object]] = None, + origin: Optional[str] = None, +) -> tuple[int, dict[str, object]]: + body = json.dumps(payload).encode("utf-8") if payload is not None else None + headers = {"X-CYD-Token": token} + if body is not None: + headers["Content-Type"] = "application/json" + if origin: + headers["Origin"] = origin + request = urllib.request.Request(url, data=body, headers=headers, method=method) + with urllib.request.urlopen(request, timeout=5) as response: + return response.status, json.loads(response.read().decode("utf-8")) + + +def main() -> int: + import sys + + sys.path.insert(0, str(SRC)) + from clean_your_data.ai_config import config_for_display, load_ai_config + import clean_your_data.gui as gui_module + from clean_your_data.gui import GuiRequestError, GuiSession, create_server + + previous_ai = os.environ.get("CLEAN_YOUR_DATA_AI_COMMAND") + os.environ["CLEAN_YOUR_DATA_AI_COMMAND"] = "cat" + try: + with tempfile.TemporaryDirectory() as tmp: + home = Path(tmp) + root = home / "workspace" + root.mkdir() + (root / "pyproject.toml").write_text("[project]\nname='demo'\n", encoding="utf-8") + (root / "README.md").write_text("# Demo\nprivate preview marker\n", encoding="utf-8") + mutable = root / "mutable.txt" + mutable.write_text("safe before scan\n", encoding="utf-8") + outside_secret = home / "outside-secret.txt" + outside_secret.write_text("must-not-follow-replacement\n", encoding="utf-8") + (root / ".env").write_text("TOKEN=must-not-be-previewed\n", encoding="utf-8") + (root / ".env.production").write_text("TOKEN=also-hidden\n", encoding="utf-8") + build = root / "build" + build.mkdir() + (build / "generated.bin").write_bytes(b"x" * 4096) + source = root / "src" + source.mkdir() + (source / "main.py").write_text("print('hello')\n", encoding="utf-8") + + trash = home / "Trash" + history = home / "state" / "cleanup-history.json" + ai_config = home / "state" / "ai-config.json" + session = GuiSession( + root, + home=home, + depth=1, + node_limit=100, + timeout=2, + time_budget=10, + trash_root=trash, + history_path=history, + ai_config_path=ai_config, + ) + + payload = session.payload() + serialized = json.dumps(payload) + assert str(home) not in serialized + assert all(not any(str(key).startswith("_") for key in node) for node in payload["nodes"]) + assert payload["scope"]["path"] == "~/workspace" + assert payload["privacy"]["preview_sent_to_ai"] is False + assert payload["scope"]["cleanup_eligible"] is False + assert "scope is protected" in payload["scope"]["cleanup_reason"] + + by_name = {node["name"]: node for node in payload["nodes"]} + readme_node = by_name["README.md"] + env_node = by_name[".env"] + production_env_node = by_name[".env.production"] + mutable_node = by_name["mutable.txt"] + build_node = by_name["build"] + + readme_preview = session.preview(readme_node["node_id"]) + assert "private preview marker" in "\n".join(readme_preview["lines"]) + secret_preview = session.preview(env_node["node_id"]) + assert "must-not-be-previewed" not in "\n".join(secret_preview["lines"]) + assert "hidden" in "\n".join(secret_preview["lines"]).lower() + production_preview = session.preview(production_env_node["node_id"]) + assert "also-hidden" not in "\n".join(production_preview["lines"]) + assert "hidden" in "\n".join(production_preview["lines"]).lower() + mutable.unlink() + mutable.symlink_to(outside_secret) + replaced_preview = session.preview(mutable_node["node_id"]) + assert "must-not-follow-replacement" not in "\n".join(replaced_preview["lines"]) + assert "unavailable" in "\n".join(replaced_preview["lines"]).lower() + try: + session.toggle_stage(env_node["node_id"]) + except GuiRequestError as exc: + assert "credential" in str(exc).lower() + else: + raise AssertionError("credential configuration must not enter the cleanup basket") + + answer = session.ask(readme_node["node_id"], "What is this?") + assert "What is this?" in answer["answer"] + assert "private preview marker" not in answer["answer"] + + os.environ.pop("CLEAN_YOUR_DATA_AI_COMMAND", None) + saved = session.set_ai_config("command", "cat") + assert saved["mode"] == "command" + assert saved["command"] == "" + assert saved["command_configured"] is True + assert load_ai_config(ai_config)["command"] == ["cat"] + displayed = config_for_display(load_ai_config(ai_config)) + assert displayed["has_api_key_field"] is False + assert displayed["stores_command_arguments"] is True + assert displayed["command"] == "" + configured_answer = session.ask(readme_node["node_id"], "Does saved config work?") + assert "Does saved config work?" in configured_answer["answer"] + assert "Category:" in configured_answer["answer"] + try: + session.set_ai_config("command", "'unterminated") + except GuiRequestError as exc: + assert "invalid AI command" in str(exc) + else: + raise AssertionError("invalid custom commands must return an actionable request error") + failing_agent = home / "failing-agent.py" + failing_agent.write_text( + "import sys\nsys.stderr.write(" + repr(str(home)) + ")\nraise SystemExit(1)\n", + encoding="utf-8", + ) + session.set_ai_config("command", f"{sys.executable} {failing_agent}") + assert str(home) not in json.dumps(session.payload()) + try: + session.ask(readme_node["node_id"], "Do not leak provider stderr") + except GuiRequestError as exc: + assert str(home) not in str(exc) + assert "did not return an answer" in str(exc) + else: + raise AssertionError("a failing provider must return a sanitized GUI error") + session.set_ai_config("command", "cat") + os.environ["CLEAN_YOUR_DATA_AI_COMMAND"] = "cat" + + staged = session.toggle_stage(build_node["node_id"]) + assert staged["staged"] is True + try: + session.move_staged_to_trash({build_node["node_id"]: "~/wrong"}) + except GuiRequestError as exc: + assert "confirmation" in str(exc).lower() + else: + raise AssertionError("mismatched path confirmation must be rejected") + assert build.exists() + + moved = session.move_staged_to_trash({build_node["node_id"]: build_node["path"]}) + assert moved["moved"] == 1 + assert not build.exists() + assert list(trash.iterdir()) + restored = session.undo() + assert restored["restored"] == 1 + assert build.exists() + + build_id = next(node["node_id"] for node in session.payload()["nodes"] if node["name"] == "build") + source_id = next(node["node_id"] for node in session.payload()["nodes"] if node["name"] == "src") + session.toggle_stage(build_id) + session.toggle_stage(source_id) + original_move_to_trash = gui_module.move_to_trash + + def fail_source_move(path, **kwargs): + if path.name == "src": + raise OSError("fixture failure with a private absolute path") + return original_move_to_trash(path, **kwargs) + + gui_module.move_to_trash = fail_source_move + try: + partial = session.move_staged_to_trash( + { + build_id: next(node["path"] for node in session.payload()["nodes"] if node["node_id"] == build_id), + source_id: next(node["path"] for node in session.payload()["nodes"] if node["node_id"] == source_id), + } + ) + finally: + gui_module.move_to_trash = original_move_to_trash + assert partial["moved"] == 1 + assert len(partial["errors"]) == 1 + assert str(home) not in json.dumps(partial["errors"]) + assert not build.exists() + assert source.exists() + assert source_id in partial["session"]["staged"] + assert session.undo()["restored"] == 1 + assert build.exists() + session.toggle_stage(source_id) + + fresh_payload = session.payload() + build_node = next(node for node in fresh_payload["nodes"] if node["name"] == "build") + backup_node = next(node for node in fresh_payload["nodes"] if node["name"] == "README.md") + session.toggle_stage(build_node["node_id"]) + session.toggle_stage(backup_node["node_id"]) + two_moves = session.move_staged_to_trash( + { + build_node["node_id"]: build_node["path"], + backup_node["node_id"]: backup_node["path"], + } + ) + assert two_moves["moved"] == 2 + original_restore = gui_module.restore_trash_record + + def fail_readme_restore(record, history_path=None): + if record.get("name") == "README.md": + raise FileExistsError("fixture failure with a private absolute path") + return original_restore(record, history_path) + + gui_module.restore_trash_record = fail_readme_restore + try: + partial_undo = session.undo() + finally: + gui_module.restore_trash_record = original_restore + assert partial_undo["restored"] == 1 + assert len(partial_undo["errors"]) == 1 + assert str(home) not in json.dumps(partial_undo["errors"]) + assert build.exists() + assert not (root / "README.md").exists() + assert len(session.last_records) == 1 + assert session.undo()["restored"] == 1 + assert (root / "README.md").exists() + + token = "gui-test-token" + server = create_server(session, token=token) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + base = f"http://127.0.0.1:{server.server_port}" + try: + with urllib.request.urlopen(base + "/", timeout=5) as response: + html = response.read().decode("utf-8") + assert token in html + assert "__CYD_SESSION_TOKEN__" not in html + + spoofed_host = urllib.request.Request(base + "/", headers={"Host": "example.com"}) + try: + urllib.request.urlopen(spoofed_host, timeout=5) + except urllib.error.HTTPError as exc: + assert exc.code == 403 + else: + raise AssertionError("a non-loopback Host header must be rejected") + + try: + urllib.request.urlopen(base + "/api/session", timeout=5) + except urllib.error.HTTPError as exc: + assert exc.code == 403 + else: + raise AssertionError("API request without the session token must fail") + + status, api_payload = request_json(base + "/api/session", token) + assert status == 200 + assert api_payload["scope"]["path"] == "~/workspace" + + try: + request_json( + base + "/api/session", + token, + origin="https://example.com", + ) + except urllib.error.HTTPError as exc: + assert exc.code == 403 + else: + raise AssertionError("cross-origin request must fail") + finally: + server.shutdown() + server.server_close() + thread.join(timeout=5) + assert not thread.is_alive() + finally: + if previous_ai is None: + os.environ.pop("CLEAN_YOUR_DATA_AI_COMMAND", None) + else: + os.environ["CLEAN_YOUR_DATA_AI_COMMAND"] = previous_ai + + print("gui test ok") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/package_test.py b/tests/package_test.py index fe10e39..45ced50 100644 --- a/tests/package_test.py +++ b/tests/package_test.py @@ -9,17 +9,24 @@ import subprocess import sys import tempfile +from importlib import resources from pathlib import Path +from typing import Optional ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" -def run_module(*args: str) -> subprocess.CompletedProcess[str]: +def run_module( + *args: str, + env_overrides: Optional[dict[str, str]] = None, +) -> subprocess.CompletedProcess[str]: env = os.environ.copy() existing = env.get("PYTHONPATH") env["PYTHONPATH"] = str(SRC) + (os.pathsep + existing if existing else "") + if env_overrides: + env.update(env_overrides) return subprocess.run( [sys.executable, "-m", "clean_your_data", *args], cwd=ROOT, @@ -35,7 +42,8 @@ def main() -> int: from clean_your_data import __version__ from clean_your_data.cli import normalize_argv - assert __version__ == "0.3.0" + assert __version__ == "0.4.0" + assert resources.files("clean_your_data").joinpath("web/index.html").is_file() assert normalize_argv([]) == ["--tui", "--path", "."] assert normalize_argv(["/tmp/project", "--focus-depth", "3"]) == [ "--tui", @@ -48,7 +56,23 @@ def main() -> int: version = run_module("--version") assert version.returncode == 0, version.stderr - assert version.stdout.strip() == "clean-your-data 0.3.0" + assert version.stdout.strip() == "clean-your-data 0.4.0" + + gui_help = run_module("gui", "--help") + assert gui_help.returncode == 0, gui_help.stderr + assert "cyd gui" in gui_help.stdout + + with tempfile.TemporaryDirectory() as state_dir: + invalid_config = run_module( + "config", + "ai", + "--command", + "'unterminated", + env_overrides={"CLEAN_YOUR_DATA_STATE_DIR": state_dir}, + ) + assert invalid_config.returncode == 2 + assert "invalid AI command" in invalid_config.stderr + assert "Traceback" not in invalid_config.stderr installed_command = shutil.which("cyd") if installed_command: @@ -59,7 +83,7 @@ def main() -> int: check=False, ) assert installed_version.returncode == 0, installed_version.stderr - assert installed_version.stdout.strip() == "clean-your-data 0.3.0" + assert installed_version.stdout.strip() == "clean-your-data 0.4.0" with tempfile.TemporaryDirectory() as tmp: home = Path(tmp) diff --git a/tests/trace_test.py b/tests/trace_test.py new file mode 100644 index 0000000..3c41b61 --- /dev/null +++ b/tests/trace_test.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Deterministic coverage for the opt-in local command tracer.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import stat +import subprocess +import sys +import tempfile +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SRC = ROOT / "src" + + +def main() -> int: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) / "project" + root.mkdir() + (root / "existing.txt").write_text("before\n", encoding="utf-8") + (root / "remove.txt").write_text("remove\n", encoding="utf-8") + state_dir = Path(tmp) / "state" + command_code = ( + "from pathlib import Path; import time; " + "root=Path.cwd(); " + "(root/'added.txt').write_text('created\\n'); " + "(root/'existing.txt').write_text('after\\n'); " + "(root/'remove.txt').unlink(); " + "time.sleep(0.08)" + ) + env = os.environ.copy() + env["PYTHONPATH"] = str(SRC) + (os.pathsep + env["PYTHONPATH"] if env.get("PYTHONPATH") else "") + result = subprocess.run( + [ + sys.executable, + "-m", + "clean_your_data", + "trace", + "--path", + str(root), + "--cwd", + str(root), + "--interval", + "0.01", + "--state-dir", + str(state_dir), + "--format", + "json", + "--", + sys.executable, + "-c", + command_code, + ], + cwd=ROOT, + env=env, + capture_output=True, + text=True, + check=False, + ) + assert result.returncode == 0, result.stderr + report = json.loads(result.stdout) + assert report["trace_schema_version"] == "1.0" + assert report["session"]["return_code"] == 0 + assert report["session"]["cwd"] == "." + assert report["observation"]["attribution"].startswith("associated with") + events = {item["path"]: item for item in report["events"]} + assert events["./added.txt"]["event_types"] == ["created"] + assert events["./existing.txt"]["event_types"] == ["modified"] + assert events["./existing.txt"]["changed_fields"] == ["size", "modified time"] + assert events["./remove.txt"]["event_types"] == ["deleted"] + assert events["./remove.txt"]["after"] is None + assert report["store"]["saved"] is True + + database = state_dir / "provenance.sqlite3" + assert database.exists() + assert stat.S_IMODE(state_dir.stat().st_mode) == 0o700 + assert stat.S_IMODE(database.stat().st_mode) == 0o600 + with sqlite3.connect(database) as connection: + session_count = connection.execute("SELECT COUNT(*) FROM trace_sessions").fetchone()[0] + event_count = connection.execute("SELECT COUNT(*) FROM trace_events").fetchone()[0] + assert session_count == 1 + assert event_count == 3 + assert str(root) not in result.stdout + + print("trace test ok") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/tui_test.py b/tests/tui_test.py index dce6de4..6c00f69 100644 --- a/tests/tui_test.py +++ b/tests/tui_test.py @@ -95,6 +95,9 @@ def main() -> int: assert state.selected()["name"] == "project" assert "metadata-only" in state.message assert [node["name"] for node in state.visible()] == ["project", "src", "lib"] + assert "3 folders" in tui.space_summary(state) + story = tui.space_story_lines(state, nodes[0], 40) + assert any("Largest visible area" in line for line, _ in story) state.selected_id = "space-grandchild-folder" assert tui.handle_key(None, state, "\n", 80) is True assert [node["name"] for node in state.visible()] == ["project", "src", "lib", "notes.txt"]