commit d9d8a4f36379d885764df004d629f6263812b1ec
Author: Alex Blank <alexblank@fastmail.com>
Date:   Thu Apr 30 18:32:49 2026 +0200

    Initial commit: modular Streamlit media viewer with Jellyfin, SSH tools, and docs

diff --git a/.env.example b/.env.example
new file mode 100644
index 0000000..76548d5
--- /dev/null
+++ b/.env.example
@@ -0,0 +1,14 @@
+JELLYFIN_URL=https://jellyfin.example.com
+JELLYFIN_API_KEY=your-api-key
+# Optional if /Users works with your API key. Otherwise set the id of the Jellyfin user whose library views should be shown.
+JELLYFIN_USER_ID=
+
+SSH_HOST=media-server.example.com
+SSH_USERNAME=username
+SSH_PORT=22
+SSH_KEY_FILENAME=/home/username/.ssh/id_rsa
+# SSH_PASSWORD=optional-password-or-key-passphrase
+REMOTE_MEDIA_ROOT=/mnt/media
+# Optional fallback prefix when REMOTE_MEDIA_ROOT mapping is not enough.
+# Example: Jellyfin gives /media/... but SSH host requires /srv/media/...
+REMOTE_PATH_PREFIX=
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..af947c9
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,41 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+.pytest_cache/
+.ruff_cache/
+.mypy_cache/
+.coverage
+htmlcov/
+
+# Packaging/build
+build/
+dist/
+*.egg-info/
+
+# Virtual environments
+.venv/
+venv/
+env/
+
+# Local configuration and secrets
+.env
+.env.*
+!.env.example
+.streamlit/secrets.toml
+.envrc
+
+# Editors/OS
+.idea/
+.vscode/
+.DS_Store
+
+# Streamlit/runtime
+.streamlit/config.toml
+
+# Local generated indexes/caches
+.cache/
+
+# Logs/temp
+*.log
+tmp/
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..3cf07d8
--- /dev/null
+++ b/README.md
@@ -0,0 +1,145 @@
+# Media Library Viewer
+
+Small Streamlit app for browsing a remote Jellyfin library, inspecting files on disk over SSH, previewing media metadata with `ffprobe`, and running safe remote job templates.
+
+See `docs/REQUIREMENTS.md` for the living requirements, decisions, and planning history.
+
+## Phase 1 features
+
+- Dashboard media counts for movies, series, and episodes, plus now-playing sessions (user, title, playback state, transcoding)
+- SQLite-indexed Media tab with full-library sort/filter for runtime, size, bitrate, explicit HDR yes/no flag, date added, codec, resolution, series, season, episode, path, and row-based selection that automatically syncs File browser to the selected item's folder
+- SSH resource overview dashboard plus detailed Resources tab for CPU, RAM, network, disk I/O, and disk space
+- Jellyfin API-key connection using `GET /Users` plus user-scoped library endpoints
+- Library selection, search, pagination, poster grid
+- Item details with Jellyfin metadata and raw JSON
+- Compact SSH remote directory browser with clickable rows, search, filters, sorting, pagination, and `[UP] ..` navigation
+- Blocking selected-file `ffprobe` preview for known video files
+- Separate container, video, audio, and subtitle metadata sections
+- `stat` inspection
+- Safe/read-only job templates, designed to extend later
+
+## Project structure
+
+```text
+.
+├── app.py                         # Thin Streamlit entrypoint
+├── pyproject.toml                 # Package metadata, dependencies, tool config
+├── requirements.txt               # Convenience install file (-e .)
+├── docs/
+│   └── REQUIREMENTS.md            # Living requirements and decision log
+├── src/
+│   └── media_library_viewer/
+│       ├── app.py                 # Thin Streamlit orchestration layer
+│       ├── config.py              # Environment/.env config
+│       ├── jobs.py                # Remote job templates
+│       ├── utils.py               # Formatting and ffprobe summaries
+│       ├── domain/                # UI-independent normalization/domain helpers
+│       ├── services/              # UI-independent application services/indexes
+│       ├── ui/                    # Streamlit UI modules split by feature area
+│       └── clients/
+│           ├── jellyfin.py        # Jellyfin API client
+│           └── ssh.py             # SSH/ffprobe client
+└── tests/
+```
+
+## Setup
+
+```bash
+python -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+```
+
+For development tools:
+
+```bash
+pip install -e '.[dev]'
+```
+
+Create `.env` or use Streamlit secrets/env vars:
+
+```bash
+JELLYFIN_URL=https://jellyfin.example.com
+JELLYFIN_API_KEY=your-api-key
+# Optional fallback if user selection via GET /Users does not work.
+JELLYFIN_USER_ID=
+
+SSH_HOST=media-server.example.com
+SSH_USERNAME=username
+SSH_PORT=22
+SSH_KEY_FILENAME=/home/username/.ssh/id_rsa
+# SSH_PASSWORD=optional-password-or-key-passphrase
+REMOTE_MEDIA_ROOT=/mnt/media
+# Optional fallback prefix when REMOTE_MEDIA_ROOT mapping is not enough.
+# Example fallback: Jellyfin /media/... -> SSH /srv/media/...
+REMOTE_PATH_PREFIX=/srv
+```
+
+Run:
+
+```bash
+streamlit run app.py
+```
+
+## Before publishing / committing
+
+Keep these out of git:
+
+- `.env` and any real secrets/tokens/passwords
+- `.streamlit/secrets.toml`
+- `.venv/`, local IDE files (`.idea/`, `.vscode/`)
+- local cache/index files under `.cache/`
+- logs and temporary files
+
+The included `.gitignore` already excludes these.
+
+## Remote server requirements
+
+For file browsing, resource metrics, and media metadata:
+
+```bash
+ffprobe -version
+python3 --version
+find --version
+stat --version
+df --version
+awk --version || true
+```
+
+The resource dashboard uses Linux `/proc`, `/sys/block`, `/bin/sh`, and a lightweight collector started over SSH. No full monitoring stack is required, but last-hour charts require the collector to have been running long enough to gather samples. Network is shown as download/upload in bytes per second, and disk I/O is shown as read/write in bytes per second. The metrics JSONL file is pruned to 7 days with a 70,000-line safety cap. SSH commands are explicitly run through `/bin/sh -c`, so the remote user's login shell may be fish or another shell. If no samples appear after 10-20 seconds, use the dashboard's `Restart` button and inspect `Collector diagnostics`.
+
+The SSH client uses your local `known_hosts` and rejects unknown host keys. Connect once manually first:
+
+```bash
+ssh user@host
+```
+
+## Extending jobs
+
+Add templates in `src/media_library_viewer/jobs.py`:
+
+```python
+JOB_TEMPLATES["my_job"] = JobTemplate(
+    name="My job",
+    description="What it does.",
+    command_template="my-command --input {path}",
+    destructive=False,
+)
+```
+
+Template variables are shell-quoted before insertion. For destructive jobs, add confirmation UI before executing.
+
+## Development checks
+
+```bash
+PYTHONPATH=src python -m py_compile app.py src/media_library_viewer/*.py src/media_library_viewer/clients/*.py
+ruff check .
+pytest
+```
+
+## Notes
+
+- Prefer Jellyfin API for library metadata.
+- Enter the Jellyfin server root URL, for example `https://jellyfin.example.com`, not `https://jellyfin.example.com/web`. The client strips a trailing `/web` defensively.
+- Prefer SSH/ffprobe for authoritative disk-level media metadata such as actual bitrate, color transfer, HDR metadata, audio channels, subtitles, and container details.
+- Keep cleanup/transcode jobs explicit and template-based; avoid free-form command execution in the UI unless this app is only used locally by trusted users.
diff --git a/app.py b/app.py
new file mode 100644
index 0000000..c4212e7
--- /dev/null
+++ b/app.py
@@ -0,0 +1,23 @@
+"""Convenience Streamlit entrypoint.
+
+The real application lives in :mod:`media_library_viewer.app` under ``src/`` so
+that the project can be packaged and reused by other frontends later. This file
+keeps the simple development command working:
+
+    streamlit run app.py
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+SRC = Path(__file__).resolve().parent / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from media_library_viewer.app import main  # noqa: E402
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/REQUIREMENTS.md b/docs/REQUIREMENTS.md
new file mode 100644
index 0000000..4eec33b
--- /dev/null
+++ b/docs/REQUIREMENTS.md
@@ -0,0 +1,316 @@
+# Media Library Viewer - Requirements and Decision Log
+
+This is a living document for the project. Update it whenever requirements, UX expectations, architecture decisions, constraints, or implementation plans change.
+
+## Product Goal
+
+Build a compact Streamlit application for browsing a remote Jellyfin media library and inspecting the corresponding media files on disk over SSH. The app should support library metadata review, direct file-system navigation, detailed media metadata inspection, and safe remote maintenance/job workflows.
+
+## Current Phase
+
+Phase 1: Jellyfin library browser plus SSH-based remote filesystem inspection and safe job templates.
+
+## Core Requirements
+
+### Jellyfin Library
+
+- Connect to a remote Jellyfin server using an API key.
+- Use the Jellyfin server root URL, not the `/web` UI URL.
+- Handle API-key auth correctly:
+  - use `GET /Users` to list available users;
+  - allow a manual `JELLYFIN_USER_ID` override;
+  - do not rely on `/Users/Me` for API-key auth.
+- List Jellyfin libraries for the selected user.
+- Browse library items with search, pagination, media type filtering, and poster cards.
+- Show item details including overview, genres, ratings where available, file path, media sources/streams, and raw Jellyfin JSON.
+- Provide a Media tab with a paginated inventory table for large libraries.
+- Media inventory should show title, series name, season, episode number, type, year, runtime, file size, bitrate, explicit HDR yes/no flag, video codec, resolution, date added, and path where available.
+- Media inventory should use Jellyfin metadata only for now; full ffprobe enrichment for all media should be deferred to a cached/background scan to avoid expensive per-item SSH probing.
+- Media inventory should support multi-library selection, type, search, full-index sort/order, HDR filter, page size, and page controls.
+- Media inventory should use a local SQLite index so sorting/filtering by nested/derived fields such as size, bitrate, HDR, codec, resolution, series, season, and episode can apply to the whole indexed library instead of only one Jellyfin page.
+- Media inventory table should be read-only; full-index sorting/filtering should be handled by the service/query layer rather than relying on frontend table sorting.
+- Media inventory should use row-based table selection and automatically sync the File browser tab to the selected row's containing directory.
+- Media row selection should update both Media selected-row state and File browser location without requiring an extra action button.
+- File browser interaction should stay explicit and simple: read-only listing plus explicit Open/Select actions rather than another row-selection grid.
+- Use valid Jellyfin `Fields` query values only, because invalid field names can cause `400 Bad Request` responses.
+
+### Remote Filesystem over SSH
+
+- Connect to a remote media server via SSH.
+- Use strict SSH host key behavior; users should connect manually once to populate `known_hosts`.
+- Browse remote directories and files rooted at a configurable default media path.
+- File browser handoff should map Jellyfin paths to `REMOTE_MEDIA_ROOT` when possible (for example `/media/...` -> `/srv/media/...` when root is `/srv/media`).
+- Support a configurable Jellyfin-to-SSH fallback path prefix for cases where `REMOTE_MEDIA_ROOT` mapping alone is not sufficient.
+- Support manual path entry and refresh.
+- Remote file listing must be compact, structured, and navigable.
+- The file table should be read-only.
+- The file table should use row selection (single-select) in an AG Grid format consistent with the Media tab.
+- The file table should not expose a visible checkbox selection column.
+- The file table should not show a visible `selected` column.
+- Include a top `[UP] ..` row, when not at `/`, to navigate to the parent directory.
+- The selected path should be visibly shown outside the table.
+- Selecting a directory row (including `[UP] ..`) should open it immediately.
+- Selecting a file row should set the selected file target for metadata/jobs.
+- Avoid emojis and hard-to-render characters in labels.
+
+### Remote Listing Controls
+
+- Show compact listing summary information:
+  - total entries;
+  - directories;
+  - files;
+  - total file size for files in the current directory.
+- Support filtering by:
+  - all entries;
+  - directories;
+  - files;
+  - file extension.
+- Support case-insensitive filename search.
+- Support sorting by:
+  - name;
+  - kind/type;
+  - size;
+  - modified time.
+- Support ascending/descending sort order.
+- Support pagination and configurable rows per page.
+
+### Disk-Level Metadata
+
+- Run `ffprobe` on selected remote files to inspect authoritative media metadata directly from disk.
+- When a known video/movie file is selected in the remote file browser, automatically run a blocking `ffprobe` preview and show a spinner while it completes.
+- Cache preview results briefly to keep repeated Streamlit reruns responsive; allow users to reload the preview manually.
+- Display `ffprobe` results in separate sections instead of one sparse all-streams table:
+  - container/format summary;
+  - video streams;
+  - audio streams;
+  - subtitle streams.
+- Video metadata should include codec, profile, resolution, pixel format, bitrate, frame rate, color range/space/transfer/primaries, side data/HDR-related metadata where available, language, title, and default flag.
+- Audio metadata should include codec, profile, channels/layout, sample rate, bitrate, language, title, default, and forced flags.
+- Subtitle metadata should include codec, language, title, default, forced, and hearing-impaired flags where available.
+- Expose raw `ffprobe` JSON for deeper inspection.
+- Keep a manual `ffprobe` action available for selected paths.
+- Support `stat` on selected paths.
+
+### Dashboard / Server Resources
+
+- Provide a dashboard tab with a compact Jellyfin media library overview and server resource overview.
+- Show Jellyfin media counts for movies, series, and series episodes on the dashboard.
+- Show currently playing Jellyfin sessions on the dashboard, including user, media title, playback state, and whether transcoding is active.
+- Provide a dashboard tab with a compact server resource overview over SSH.
+- Provide a separate Resources tab for detailed resource charts, collector controls, diagnostics, and raw samples.
+- Show CPU and RAM usage for the last hour.
+- Show average and spike/peak values for network throughput and disk I/O.
+- Show used, available, and total disk space for the configured media root, falling back to `/`.
+- Use a lightweight remote collector that reads Linux `/proc`, `/sys/block`, and `df` data into a JSONL file under `/tmp`.
+- The collector should rotate/prune its JSONL metrics file so it does not grow unbounded; default retention is 7 days with a 70,000-line safety cap.
+- The collector should be startable/stoppable/restartable from the dashboard and should not require installing a full monitoring stack.
+- Last-hour charts require the collector to have been running long enough to collect samples.
+- Network throughput should be shown split into down/download and up/upload.
+- Network throughput should use bytes-per-second display units such as KB/s, MB/s, and GB/s to avoid bit/byte ambiguity.
+- Disk throughput should be shown split into read and write.
+- Network and disk throughput charts should scale values into readable units such as KB/s, MB/s, and GB/s.
+
+### Remote Jobs
+
+- Support running remote jobs over SSH using explicit templates.
+- Phase 1 jobs should be safe/read-only by default.
+- Avoid arbitrary free-form command execution in the UI.
+- Job templates should be centralized in `jobs.py` for future extension.
+- Command template values must be shell-quoted before execution.
+- Future destructive jobs should require explicit confirmation.
+
+## Current Architecture
+
+- `app.py` - Thin root Streamlit entrypoint for `streamlit run app.py`.
+- `pyproject.toml` - Authoritative package metadata, dependencies, and tool configuration.
+- `requirements.txt` - Convenience install file that installs the local package editable.
+- `src/media_library_viewer/app.py` - Thin Streamlit orchestration layer.
+- `src/media_library_viewer/ui/` - Streamlit UI modules split by feature area (dashboard, media, file browser, library, preview/tools).
+- `src/media_library_viewer/clients/jellyfin.py` - Jellyfin API wrapper.
+- `src/media_library_viewer/clients/ssh.py` - SSH command execution, directory listing, `stat`, and `ffprobe` helpers.
+- `src/media_library_viewer/domain/` - UI-independent normalization/domain helpers.
+- `src/media_library_viewer/services/` - UI-independent application services such as the SQLite media index.
+- `src/media_library_viewer/jobs.py` - Remote job template definitions and runner.
+- `src/media_library_viewer/utils.py` - Formatting and media metadata summarization helpers.
+- `src/media_library_viewer/config.py` - Environment variable and `.env` configuration loading.
+- `docs/REQUIREMENTS.md` - Living requirements and decision log.
+- `tests/` - Reserved for future test coverage.
+
+## Key Implementation Decisions
+
+- Prefer the Jellyfin API for library and server metadata.
+- Prefer SSH plus `ffprobe` for disk-authoritative stream/container metadata.
+- Use API-key auth for Jellyfin, but select a user explicitly for user-scoped endpoints.
+- Use `streamlit-aggrid` as a required dependency for Media table row selection. Avoid optional frontend fallbacks that create multiple interaction models.
+- Keep remote jobs template-based to reduce accidental destructive actions.
+- Keep the Phase 1 UI compact and structured rather than using large per-row buttons.
+- Use a `src/` package layout so the project can grow without accumulating many root-level modules.
+- Keep root `app.py` as a compatibility/convenience wrapper for Streamlit.
+- Keep clients, domain normalization, and application services independent from Streamlit so the frontend can later be replaced by React/FastAPI or another UI.
+- Keep Streamlit rendering split into small UI modules so interaction bugs can be debugged in feature-local code instead of one monolithic app file.
+
+## Security and Safety Requirements
+
+- Do not hardcode secrets.
+- Use `.env`, environment variables, or Streamlit secrets for credentials.
+- Keep `.env` and Streamlit secrets out of version control.
+- Reject unknown SSH host keys by default.
+- Treat SSH jobs as potentially dangerous and keep them explicit/template-based.
+- Add confirmation steps before implementing cleanup, delete, transcode-replace, or other destructive workflows.
+
+## Configuration Requirements
+
+Supported environment variables:
+
+```bash
+JELLYFIN_URL=
+JELLYFIN_API_KEY=
+JELLYFIN_USER_ID=
+SSH_HOST=
+SSH_USERNAME=
+SSH_PORT=22
+SSH_KEY_FILENAME=
+SSH_PASSWORD=
+REMOTE_MEDIA_ROOT=
+REMOTE_PATH_PREFIX=
+```
+
+## Known External Requirements
+
+Remote server should have:
+
+- Linux `/proc` and `/sys/block` for resource metrics
+- `/bin/sh` for POSIX command execution, even when the user's login shell is fish or another non-POSIX shell
+- POSIX shell utilities including `awk`, `date`, `tail`, `df`, `kill`, and `nohup`
+- `python3`
+- GNU/coreutils-compatible `find` and `stat`
+- `ffprobe` for media metadata inspection
+
+Local app dependencies are declared in `pyproject.toml`; `requirements.txt` installs the package editable for convenience. Runtime dependencies include:
+
+- `streamlit`
+- `streamlit-aggrid`
+- `requests`
+- `paramiko`
+- `python-dotenv`
+- `pandas`
+
+## Backlog / Future Extensions
+
+- Add transcode job templates.
+- Add cleanup job templates with dry-run and explicit confirmation.
+- Add subtitle/audio-track diagnostics.
+- Add sidecar file inspection for `.nfo`, `.srt`, images, and metadata files.
+- Compare Jellyfin metadata against disk metadata and sidecars.
+- Add long-running job tracking/log streaming.
+- Add saved presets for common media roots and job templates.
+- Add file previews for text sidecars.
+- Add richer HDR/Dolby Vision/bit-depth summaries from `ffprobe`.
+- Add optional integration with existing monitoring stacks such as Prometheus/node_exporter, Netdata, or sysstat/sar.
+
+## Decision Log
+
+### 2026-04-30 - Initial app plan
+
+- Planned a Streamlit app that uses the Jellyfin API as the primary metadata source.
+- Decided SSH should be used for disk inspection and future maintenance jobs.
+
+### 2026-04-30 - Phase 1 implementation
+
+- Created the initial app structure with Jellyfin, SSH, jobs, config, and utility modules.
+- Added safe/read-only remote job templates.
+- Added `ffprobe` and `stat` inspection.
+
+### 2026-04-30 - Jellyfin API fixes
+
+- Replaced `/Users/Me` usage with `GET /Users` plus user selection.
+- Added `JELLYFIN_USER_ID` override.
+- Cleaned Jellyfin `Fields` values to avoid 400 responses.
+- Added defensive stripping of trailing `/web` from Jellyfin URLs.
+
+### 2026-04-30 - Remote file browser evolution
+
+- Added interactive remote listing.
+- Removed emoji and hard-to-render characters.
+- Added search, filtering, sorting, pagination, and compact listing summary.
+- Reworked listing from large button rows into a compact table.
+- Switched to `streamlit-aggrid` for file-browser-like row click behavior.
+- Removed visible checkbox/selection column behavior.
+- Added `[UP] ..` top row for parent directory navigation.
+
+### 2026-04-30 - Selected-file metadata preview
+
+- Added a requirement for automatic `ffprobe` preview when known video files are selected.
+- Initially explored asynchronous/non-blocking preview, then changed to a blocking call with a spinner because it is more streamlined for this app.
+- Decided to cache preview results briefly and provide a manual reload action.
+- Decided `ffprobe` output should be separated into container, video, audio, and subtitle sections to avoid sparse mixed-stream tables.
+
+### 2026-04-30 - Process requirement
+
+- Added this living requirements and decision log document.
+- Added a global agent skill to encourage maintaining such a document for future projects.
+
+### 2026-04-30 - Repository restructuring
+
+- Restructured the project into a larger-project-ready `src/media_library_viewer/` package layout.
+- Kept a thin root `app.py` entrypoint so `streamlit run app.py` remains the primary launch command.
+- Moved service clients into `src/media_library_viewer/clients/`.
+- Added `pyproject.toml` with runtime dependencies, development extras, Ruff configuration, and pytest configuration.
+- Simplified `requirements.txt` to install the local project editable.
+- Expanded `.gitignore` for Python caches, build artifacts, virtual environments, local secrets, editor files, and logs.
+
+### 2026-04-30 - Resource dashboard
+
+- Added a dashboard requirement for CPU, RAM, network, disk I/O, and disk space overview.
+- Decided that true last-hour metrics require collection over time; implemented a lightweight SSH-started remote collector instead of requiring Prometheus, Netdata, or sysstat.
+- The collector stores JSONL samples in `/tmp` and can be started/stopped from the dashboard.
+- Charts show the last hour of collected samples; the dashboard becomes more useful once the collector has been running for a while.
+- Last-hour filtering uses epoch seconds rather than local naive datetimes to avoid timezone-offset issues between the app host and remote sample timestamps.
+- Fixed SSH command execution to explicitly use `/bin/sh -c` so POSIX resource commands work even when the remote user's login shell is fish.
+- Added explicit Streamlit keys to dashboard/file/tool buttons to avoid duplicate auto-generated element IDs as the UI grows.
+- Changed the resource collector script from bash-specific syntax to POSIX `/bin/sh` syntax and added dashboard diagnostics/restart controls for collector troubleshooting.
+- Added 7-day metrics file pruning plus a 70,000-line safety cap to prevent the JSONL file from growing without bound.
+- Split network charts and metrics into download and upload, and disk charts and metrics into read and write.
+- Changed network display units from bits per second to bytes per second to avoid Kbps/KB/s ambiguity; the collector still stores bit-rate compatibility fields for old/debug consumers.
+- Scaled network and disk throughput charts into readable units such as KB/s, MB/s, and GB/s instead of plotting raw base units.
+- Updated collector startup to remove old temporary metrics/log files when a new collector process is started after a schema/display change.
+- Moved detailed resource charts, raw samples, diagnostics, and collector controls into a dedicated Resources tab; the Dashboard now keeps a compact overview.
+- Removed the CPU/RAM chart from the Dashboard and kept detailed charts in the Resources tab.
+- Renamed the remote files tab to File browser.
+- Added Jellyfin media counts for movies, series, and episodes to the Dashboard using lightweight count queries.
+- Added a Dashboard now-playing section sourced from Jellyfin sessions, showing who is currently playing what and whether each session is transcoding.
+
+### 2026-04-30 - Media inventory tab
+
+- Added a paginated Media tab for file-oriented Jellyfin metadata.
+- Decided not to fetch all media at once because large libraries can make API responses and Streamlit rendering slow.
+- Decided to derive length, size, bitrate, HDR flag, date added, codec, and resolution from Jellyfin metadata for now.
+- Added series name, season, and episode number for episode rows.
+- Added server-side sort/order controls and read-only AG Grid column sorting/filtering for the loaded page.
+- Reworked the Media tab to use a local SQLite media index for full-library sorting/filtering, including numeric sorting for size and bitrate.
+- Added last index build duration metadata to the Media tab status line.
+- Replaced single-library selection with multi-library selection so users can include/exclude multiple libraries in the indexed table.
+- Changed HDR display from blank/no-value to explicit yes/no.
+- Added row selection plus an Open folder action in the Media tab that sets the File browser to the containing folder.
+- Restored row-based Media table selection while keeping File browser state changes limited to the explicit Open folder button.
+- Updated Media tab behavior so selecting a row automatically syncs the File browser folder to that item's containing directory; removed the extra Open folder button step.
+- Restored File browser table row selection with AG Grid (single-select), using a table interaction style consistent with the Media tab.
+- Reintroduced open-on-select behavior in File browser: selecting a directory row (including `[UP] ..`) opens it immediately, while file rows update selected target path.
+- Refined Media table column presentation with explicit user-friendly headers and null-safe display formatting to keep the grid readable and consistent.
+- Added broad inline/module documentation across clients, domain, services, and Streamlit adapter modules to make debugging and future frontend extraction easier.
+- Simplified the File browser by removing its interactive AG Grid and using a read-only listing with explicit Open/Select controls, reducing cross-tab state interactions with the Media grid.
+- Removed optional/compatibility code paths around the Media table grid and old file-browser state aliases to keep the interaction model easier to reason about during debugging.
+- Split the Streamlit frontend into dedicated UI modules (dashboard, media, file browser, library, preview/tools) and reduced `app.py` to orchestration glue.
+- Reviewed remote path handling and kept shell interactions routed through quoted paths (`shlex.quote`) while UI/path-parent operations use POSIX path handling, preserving paths with spaces.
+- Fixed file browser handoff/navigation to reset stale search/filter/page state when changing folders, preventing old filters from hiding all entries in the newly opened folder.
+- Reworked file browser state to separate current directory from selected path. Selecting a file no longer changes the directory being listed, while opening a folder updates the current directory and keeps path input synchronized.
+- Made remote directory listing fail explicitly when the current path is not a directory and recover by listing the parent, preventing file paths from appearing as empty directories.
+- Made File browser Refresh/Select folder apply a manually typed path if it differs from the current folder, reducing confusion when manually navigating.
+- Moved media normalization into `domain/media.py` and index/query logic into `services/media_index.py` to make the project less Streamlit-specific and easier to expose through a future API/React frontend.
+- Deferred full ffprobe enrichment for every item to a future cached/background scan.
+- Fixed network byte parsing to split `/proc/net/dev` lines at the colon first, so interface indentation differences do not shift fields and accidentally report packet counts instead of byte counts.
+- Fixed a follow-up `/proc/net/dev` parsing issue where leading whitespace after the colon could produce an empty first split field in some `awk` implementations, resulting in zero network rates. Added `/proc/net/dev` snapshots to collector diagnostics.
+- Simplified File browser directory error behavior: stopped automatic parent-directory fallback and now show the direct listing error for the current path.
+- Added configurable `REMOTE_PATH_PREFIX` support so Jellyfin paths can be mapped to SSH-visible paths when opening folders from Media/Library tabs (for example `/media/...` -> `/srv/media/...`).
+- Updated path handoff logic to prefer mapping through `REMOTE_MEDIA_ROOT` (anchor replacement using the root basename, e.g. `media`) and use `REMOTE_PATH_PREFIX` as fallback.
+- Added a public-repo readiness note in README describing what local/sensitive files must stay out of version control.
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..ac424ff
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,42 @@
+[build-system]
+requires = ["hatchling>=1.24"]
+build-backend = "hatchling.build"
+
+[project]
+name = "media-library-viewer"
+version = "0.1.0"
+description = "Streamlit app for browsing Jellyfin libraries and inspecting remote media files over SSH."
+readme = "README.md"
+requires-python = ">=3.11"
+license = { text = "MIT" }
+authors = [{ name = "Media Library Viewer contributors" }]
+dependencies = [
+    "streamlit>=1.35",
+    "streamlit-aggrid>=1.0",
+    "requests>=2.31",
+    "paramiko>=3.4",
+    "python-dotenv>=1.0",
+    "pandas>=2.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "ruff>=0.4",
+    "pytest>=8.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/media_library_viewer"]
+
+[tool.ruff]
+line-length = 120
+target-version = "py311"
+src = ["src", "tests"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]
+ignore = ["E501"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["src"]
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..d6e1198
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1 @@
+-e .
diff --git a/src/media_library_viewer/__init__.py b/src/media_library_viewer/__init__.py
new file mode 100644
index 0000000..77edd60
--- /dev/null
+++ b/src/media_library_viewer/__init__.py
@@ -0,0 +1,4 @@
+"""Media Library Viewer package."""
+
+__all__ = ["__version__"]
+__version__ = "0.1.0"
diff --git a/src/media_library_viewer/app.py b/src/media_library_viewer/app.py
new file mode 100644
index 0000000..428ae3a
--- /dev/null
+++ b/src/media_library_viewer/app.py
@@ -0,0 +1,339 @@
+"""Streamlit UI adapter for Media Library Viewer.
+
+This module is intentionally thin. Reusable logic lives in:
+
+- ``clients/`` for Jellyfin, SSH, and resource collection integrations.
+- ``domain/`` for UI-independent normalization of Jellyfin/media metadata.
+- ``services/`` for application services such as the SQLite media index.
+- ``ui/`` for Streamlit rendering grouped by feature area.
+
+Keeping those boundaries makes it easier to replace this Streamlit frontend with
+another frontend later while preserving the backend/domain code.
+"""
+
+from __future__ import annotations
+
+import json
+import posixpath
+from pathlib import PurePosixPath
+from typing import Any
+
+import streamlit as st
+
+from media_library_viewer.clients.jellyfin import JellyfinClient
+from media_library_viewer.clients.ssh import RemoteSSHClient
+from media_library_viewer.config import load_config
+from media_library_viewer.ui.dashboard import render_media_overview, render_now_playing, render_resource_dashboard
+from media_library_viewer.ui.file_browser import render_file_browser, set_file_browser_path
+from media_library_viewer.ui.library import show_item_card, show_item_detail
+from media_library_viewer.ui.media import render_media_tab
+from media_library_viewer.ui.preview import render_ssh_tools
+
+st.set_page_config(page_title="Media Library Viewer", layout="wide")
+
+
+@st.cache_resource(show_spinner=False)
+def get_jellyfin_client(base_url: str, api_key: str) -> JellyfinClient:
+    """Return a cached Jellyfin client for the current server/API-key pair."""
+    return JellyfinClient(base_url, api_key)
+
+
+@st.cache_data(ttl=300, show_spinner=False)
+def cached_users(base_url: str, api_key: str):
+    """Cache the Jellyfin user list to avoid repeated auth-scoped requests."""
+    return get_jellyfin_client(base_url, api_key).users()
+
+
+@st.cache_resource(show_spinner=False)
+def get_ssh_client(host: str, username: str, port: int, key_filename: str, password: str) -> RemoteSSHClient:
+    """Return a cached SSH connection wrapper for the active remote server."""
+    client = RemoteSSHClient(
+        host=host,
+        username=username,
+        port=port,
+        key_filename=key_filename or None,
+        password=password or None,
+    )
+    client.connect()
+    return client
+
+
+@st.cache_data(ttl=60, show_spinner=False)
+def cached_libraries(base_url: str, api_key: str, user_id: str):
+    """Cache library views for the selected Jellyfin user."""
+    return get_jellyfin_client(base_url, api_key).libraries(user_id)
+
+
+@st.cache_data(ttl=60, show_spinner=False)
+def cached_items(
+    base_url: str,
+    api_key: str,
+    user_id: str,
+    parent_id: str | None,
+    search: str,
+    media_types: str,
+    start: int,
+    limit: int,
+    sort_by: str = "SortName",
+    sort_order: str = "Ascending",
+):
+    """Cache paginated Jellyfin item list requests."""
+    return get_jellyfin_client(base_url, api_key).items(
+        user_id=user_id,
+        parent_id=parent_id,
+        search=search,
+        include_item_types=media_types,
+        start_index=start,
+        limit=limit,
+        sort_by=sort_by,
+        sort_order=sort_order,
+    )
+
+
+@st.cache_data(ttl=60, show_spinner=False)
+def cached_item(base_url: str, api_key: str, user_id: str, item_id: str):
+    """Cache a single Jellyfin item detail payload."""
+    return get_jellyfin_client(base_url, api_key).item(user_id, item_id)
+
+
+@st.cache_data(ttl=300, show_spinner=False)
+def cached_media_counts(base_url: str, api_key: str, user_id: str):
+    """Cache dashboard-level media counts for movies/series/episodes."""
+    return get_jellyfin_client(base_url, api_key).media_counts(user_id)
+
+
+@st.cache_data(ttl=15, show_spinner=False)
+def cached_active_sessions(base_url: str, api_key: str):
+    """Cache active Jellyfin sessions briefly for dashboard now-playing status."""
+    return get_jellyfin_client(base_url, api_key).active_sessions()
+
+
+@st.cache_data(ttl=30, show_spinner=False)
+def cached_dir_listing(host: str, username: str, port: int, key_filename: str, password: str, path: str):
+    """Cache remote directory listings briefly for snappier browsing."""
+    ssh = get_ssh_client(host, username, port, key_filename, password)
+    result = ssh.list_dir(path)
+    if result.exit_status != 0:
+        raise RuntimeError(result.stderr or result.stdout)
+    return json.loads(result.stdout)
+
+
+@st.cache_data(ttl=300, show_spinner=False)
+def cached_ffprobe_preview(host: str, username: str, port: int, key_filename: str, password: str, path: str):
+    """Cache selected-file ffprobe previews so repeated reruns stay responsive."""
+    ssh = get_ssh_client(host, username, port, key_filename, password)
+    return ssh.ffprobe_json(path)
+
+
+def apply_remote_path_prefix(path: str, prefix: str) -> str:
+    """Apply an optional fallback prefix for Jellyfin->SSH path handoff."""
+    if not path:
+        return path
+    normalized_prefix = (prefix or "").strip()
+    if not normalized_prefix:
+        return path
+    normalized_prefix = normalized_prefix.rstrip("/")
+    if path == normalized_prefix or path.startswith(normalized_prefix + "/"):
+        return posixpath.normpath(path)
+    if path.startswith("/"):
+        return posixpath.normpath(normalized_prefix + path)
+    return posixpath.normpath(posixpath.join(normalized_prefix, path))
+
+
+def map_path_to_media_root(path: str, media_root: str) -> str:
+    """Map a Jellyfin path to the configured SSH media root when possible.
+
+    Example:
+    - path: ``/media/shows/Show/E01.mkv``
+    - media_root: ``/srv/media``
+    - result: ``/srv/media/shows/Show/E01.mkv``
+
+    If the path is already under ``media_root``, it is returned unchanged.
+    If the final segment of ``media_root`` (e.g. ``media``) appears in the
+    Jellyfin path, the prefix up to that segment is replaced by ``media_root``.
+    """
+    if not path:
+        return path
+    normalized_root = (media_root or "").strip()
+    if not normalized_root:
+        return path
+    normalized_root = posixpath.normpath(normalized_root)
+
+    raw_parts = [part for part in str(path).split("/") if part]
+    if not raw_parts:
+        return path
+
+    path_absolute = "/" + "/".join(raw_parts)
+    if path_absolute == normalized_root or path_absolute.startswith(normalized_root + "/"):
+        return path_absolute
+
+    root_anchor = posixpath.basename(normalized_root)
+    if not root_anchor:
+        return path
+
+    if root_anchor in raw_parts:
+        anchor_index = raw_parts.index(root_anchor)
+        remainder_parts = raw_parts[anchor_index + 1 :]
+        return posixpath.join(normalized_root, *remainder_parts) if remainder_parts else normalized_root
+
+    return path
+
+
+def resolve_remote_media_path(path: str, media_root: str, fallback_prefix: str) -> str:
+    """Resolve Jellyfin paths to SSH-visible paths.
+
+    Strategy:
+    1. Prefer mapping to ``REMOTE_MEDIA_ROOT`` when it can anchor on the root
+       basename (e.g. ``media`` in ``/srv/media``).
+    2. If no mapping happened, apply optional fallback prefix.
+    """
+    if not path:
+        return path
+    mapped = map_path_to_media_root(path, media_root)
+    if mapped and mapped != path:
+        return mapped
+    return apply_remote_path_prefix(mapped or path, fallback_prefix)
+
+
+def credentials_panel():
+    """Render connection settings and return normalized Jellyfin/SSH inputs."""
+    cfg = load_config()
+    with st.sidebar:
+        st.header("Connections")
+        with st.expander("Jellyfin", expanded=True):
+            jellyfin_url = st.text_input("URL", value=cfg.jellyfin.url, placeholder="https://jellyfin.example.com")
+            jellyfin_api_key = st.text_input("API key", value=cfg.jellyfin.api_key, type="password")
+            jellyfin_user_id = st.text_input(
+                "User ID override",
+                value=cfg.jellyfin.user_id,
+                help="Optional. API keys are not user sessions, so the app normally lists users with /Users and lets you choose one.",
+            )
+        with st.expander("SSH", expanded=True):
+            ssh_host = st.text_input("Host", value=cfg.ssh.host)
+            ssh_username = st.text_input("Username", value=cfg.ssh.username)
+            ssh_port = st.number_input("Port", min_value=1, max_value=65535, value=cfg.ssh.port)
+            ssh_key = st.text_input("Private key path", value=cfg.ssh.key_filename)
+            ssh_password = st.text_input("Password / passphrase", value=cfg.ssh.password, type="password")
+            media_root = st.text_input("Default media root", value=cfg.ssh.media_root, placeholder="/mnt/media")
+            remote_path_prefix = st.text_input(
+                "Path prefix for Jellyfin paths",
+                value=cfg.ssh.path_prefix,
+                placeholder="/srv",
+                help="Fallback prefix when REMOTE_MEDIA_ROOT mapping is not enough.",
+            )
+    return jellyfin_url, jellyfin_api_key, jellyfin_user_id, ssh_host, ssh_username, int(ssh_port), ssh_key, ssh_password, media_root, remote_path_prefix
+
+
+def main():
+    """Application entrypoint used by the root ``app.py`` wrapper."""
+    st.title("Media Library Viewer")
+    st.caption("Phase 1: Jellyfin browser + SSH filesystem inspection + safe remote job templates")
+
+    jellyfin_url, jellyfin_api_key, jellyfin_user_id, ssh_host, ssh_username, ssh_port, ssh_key, ssh_password, media_root, remote_path_prefix = credentials_panel()
+    if not jellyfin_url or not jellyfin_api_key:
+        st.info("Enter Jellyfin connection details in the sidebar.")
+        return
+
+    try:
+        client = get_jellyfin_client(jellyfin_url, jellyfin_api_key)
+        users = cached_users(jellyfin_url, jellyfin_api_key)
+    except Exception as exc:
+        st.error(f"Jellyfin connection failed: {exc}")
+        st.caption("Note: this app uses an API key with GET /Users. /Users/Me is only reliable for user access tokens, not server API keys.")
+        return
+
+    if jellyfin_user_id:
+        user_id = jellyfin_user_id
+    else:
+        if not users:
+            st.error("No Jellyfin users returned from /Users. Set JELLYFIN_USER_ID manually.")
+            return
+        user_options = {f"{user.get('Name', 'Unnamed')} ({user['Id']})": user["Id"] for user in users}
+        with st.sidebar:
+            selected_user = st.selectbox("Jellyfin user", list(user_options.keys()))
+        user_id = user_options[selected_user]
+
+    tab_dashboard, tab_resources, tab_media, tab_library, tab_files = st.tabs(
+        ["Dashboard", "Resources", "Media", "Jellyfin library", "File browser"]
+    )
+
+    with tab_dashboard:
+        render_media_overview(cached_media_counts, jellyfin_url, jellyfin_api_key, user_id)
+        st.divider()
+        render_now_playing(cached_active_sessions, jellyfin_url, jellyfin_api_key)
+        st.divider()
+        if not ssh_host or not ssh_username:
+            st.info("Enter SSH connection details in the sidebar for server resource overview.")
+        else:
+            ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
+            render_resource_dashboard(get_ssh_client, ssh_args, media_root or "/", detailed=False)
+
+    with tab_resources:
+        if not ssh_host or not ssh_username:
+            st.info("Enter SSH connection details in the sidebar.")
+        else:
+            ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
+            render_resource_dashboard(get_ssh_client, ssh_args, media_root or "/", detailed=True)
+
+    libraries = cached_libraries(jellyfin_url, jellyfin_api_key, user_id)
+
+    def set_prefixed_file_browser_path(path: str, selected_path: str | None = None, reset_filters: bool = True) -> None:
+        set_file_browser_path(
+            resolve_remote_media_path(path, media_root, remote_path_prefix),
+            resolve_remote_media_path(selected_path, media_root, remote_path_prefix) if selected_path else None,
+            reset_filters,
+        )
+
+    with tab_media:
+        render_media_tab(client, user_id, libraries, set_prefixed_file_browser_path)
+
+    with tab_library:
+        if not libraries:
+            st.warning("No libraries found.")
+            return
+        lib_by_name = {lib["Name"]: lib for lib in libraries}
+        with st.sidebar:
+            st.header("Library filters")
+            lib_name = st.selectbox("Library", list(lib_by_name.keys()))
+            media_types = st.multiselect("Media types", ["Movie", "Series", "Episode", "Video", "Audio"], default=[])
+            search = st.text_input("Search")
+            limit = st.slider("Items per page", 10, 200, 50, step=10)
+            page = st.number_input("Page", min_value=1, value=1)
+
+        response = cached_items(
+            jellyfin_url,
+            jellyfin_api_key,
+            user_id,
+            lib_by_name[lib_name]["Id"],
+            search,
+            ",".join(media_types),
+            (page - 1) * limit,
+            limit,
+        )
+        items = response.get("Items", [])
+        st.subheader(f"{lib_name} ({response.get('TotalRecordCount', len(items))} items)")
+        cols = st.columns(5)
+        for idx, item in enumerate(items):
+            with cols[idx % 5]:
+                show_item_card(client, item)
+
+        if st.session_state.get("selected_item_id"):
+            st.divider()
+            detail = cached_item(jellyfin_url, jellyfin_api_key, user_id, st.session_state["selected_item_id"])
+            show_item_detail(client, detail)
+            if detail.get("Path") and st.button("Open item path in file browser", key="library_open_item_path_in_file_browser"):
+                resolved_path = resolve_remote_media_path(detail["Path"], media_root, remote_path_prefix)
+                set_file_browser_path(str(PurePosixPath(resolved_path).parent), resolved_path)
+
+    with tab_files:
+        if not ssh_host or not ssh_username:
+            st.info("Enter SSH connection details in the sidebar.")
+            return
+        ssh_args = (ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
+        selected_path = render_file_browser(cached_dir_listing, ssh_args, media_root or "/")
+        ssh = get_ssh_client(ssh_host, ssh_username, ssh_port, ssh_key, ssh_password)
+        render_ssh_tools(ssh, ssh_args, selected_path, cached_ffprobe_preview)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/src/media_library_viewer/clients/__init__.py b/src/media_library_viewer/clients/__init__.py
new file mode 100644
index 0000000..e09d88b
--- /dev/null
+++ b/src/media_library_viewer/clients/__init__.py
@@ -0,0 +1,6 @@
+"""External service clients.
+
+Modules in this package talk to systems outside the app: Jellyfin over HTTP,
+the media server over SSH, and the lightweight remote resource collector.
+They should not import Streamlit.
+"""
diff --git a/src/media_library_viewer/clients/jellyfin.py b/src/media_library_viewer/clients/jellyfin.py
new file mode 100644
index 0000000..d58b2f2
--- /dev/null
+++ b/src/media_library_viewer/clients/jellyfin.py
@@ -0,0 +1,147 @@
+"""Jellyfin HTTP API client.
+
+This module is deliberately independent from Streamlit. It wraps only the API
+calls the app currently needs and returns plain Python dictionaries/lists so a
+future FastAPI/React frontend can reuse the same client.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import requests
+
+
+# Jellyfin validates Fields against its ItemFields enum. Keep this list to
+# documented/commonly supported optional fields; invalid names cause 400s.
+DEFAULT_FIELDS = ",".join(
+    [
+        "DateCreated",
+        "Genres",
+        "MediaSources",
+        "Overview",
+        "Path",
+        "People",
+        "PremiereDate",
+        "ProviderIds",
+        "Tags",
+    ]
+)
+
+
+class JellyfinClient:
+    """Small wrapper around the Jellyfin/Emby-compatible HTTP API."""
+
+    def __init__(self, base_url: str, api_key: str, timeout: int = 30):
+        if not base_url:
+            raise ValueError("Jellyfin URL is required")
+        if not api_key:
+            raise ValueError("Jellyfin API key is required")
+
+        # Use the server root, not the web UI path. Users often paste
+        # https://host/web; API endpoints live at https://host/...
+        self.base_url = base_url.rstrip("/")
+        if self.base_url.endswith("/web"):
+            self.base_url = self.base_url[:-4]
+        self.api_key = api_key
+        self.timeout = timeout
+        self.session = requests.Session()
+        self.session.headers.update(
+            {
+                "X-Emby-Token": api_key,
+                "Accept": "application/json",
+                "X-Emby-Authorization": 'MediaBrowser Client="MediaLibraryViewer", Device="Streamlit", DeviceId="streamlit", Version="0.1"',
+            }
+        )
+
+    def get(self, path: str, **params: Any) -> dict[str, Any]:
+        """GET a Jellyfin endpoint and include useful response text on errors."""
+        clean_params = {k: v for k, v in params.items() if v is not None and v != ""}
+        response = self.session.get(
+            f"{self.base_url}{path}", params=clean_params, timeout=self.timeout
+        )
+        try:
+            response.raise_for_status()
+        except requests.HTTPError as exc:
+            detail = response.text[:500]
+            raise requests.HTTPError(
+                f"{response.status_code} for {response.url}: {detail}",
+                response=response,
+            ) from exc
+        return response.json()
+
+    def users(self) -> list[dict[str, Any]]:
+        """List users visible to this API key.
+
+        Jellyfin API keys are server-level tokens, not user session tokens, so
+        /Users/Me often fails with API-key auth. The user id selected here is
+        then used for user-scoped library endpoints.
+        """
+        return self.get("/Users")
+
+    def libraries(self, user_id: str) -> list[dict[str, Any]]:
+        """Return top-level library views visible to the selected Jellyfin user."""
+        return self.get(f"/Users/{user_id}/Views").get("Items", [])
+
+    def items(
+        self,
+        user_id: str,
+        parent_id: str | None = None,
+        start_index: int = 0,
+        limit: int = 50,
+        search: str | None = None,
+        include_item_types: str | None = None,
+        recursive: bool = True,
+        sort_by: str = "SortName",
+        sort_order: str = "Ascending",
+    ) -> dict[str, Any]:
+        """Return a paginated item list for a user/library.
+
+        This is used by both the visual library browser and the media-index
+        builder. Keep arguments close to Jellyfin's own query parameters so the
+        service layer can request server-side pagination and basic sorting.
+        """
+        return self.get(
+            f"/Users/{user_id}/Items",
+            ParentId=parent_id,
+            StartIndex=start_index,
+            Limit=limit,
+            SearchTerm=search,
+            IncludeItemTypes=include_item_types,
+            Recursive=str(recursive).lower(),
+            Fields=DEFAULT_FIELDS,
+            SortBy=sort_by,
+            SortOrder=sort_order,
+        )
+
+    def item_count(self, user_id: str, include_item_types: str, parent_id: str | None = None) -> int:
+        """Return a count using Jellyfin's TotalRecordCount without fetching rows."""
+        response = self.get(
+            f"/Users/{user_id}/Items",
+            ParentId=parent_id,
+            Recursive="true",
+            IncludeItemTypes=include_item_types,
+            Limit=0,
+        )
+        return int(response.get("TotalRecordCount", 0))
+
+    def media_counts(self, user_id: str) -> dict[str, int]:
+        """Return dashboard-level counts for the main media types."""
+        return {
+            "movies": self.item_count(user_id, "Movie"),
+            "series": self.item_count(user_id, "Series"),
+            "episodes": self.item_count(user_id, "Episode"),
+        }
+
+    def item(self, user_id: str, item_id: str) -> dict[str, Any]:
+        return self.get(f"/Users/{user_id}/Items/{item_id}", Fields=DEFAULT_FIELDS)
+
+    def active_sessions(self, active_within_seconds: int = 300) -> list[dict[str, Any]]:
+        """Return currently active sessions that have a now-playing item."""
+        payload = self.get("/Sessions", ActiveWithinSeconds=active_within_seconds)
+        sessions = payload if isinstance(payload, list) else []
+        return [session for session in sessions if session.get("NowPlayingItem")]
+
+    def image_url(self, item_id: str, image_type: str = "Primary") -> str:
+        """Build an authenticated image URL suitable for st.image/browser use."""
+        return f"{self.base_url}/Items/{item_id}/Images/{image_type}?api_key={self.api_key}"
diff --git a/src/media_library_viewer/clients/resources.py b/src/media_library_viewer/clients/resources.py
new file mode 100644
index 0000000..c8e6d30
--- /dev/null
+++ b/src/media_library_viewer/clients/resources.py
@@ -0,0 +1,311 @@
+"""Remote resource collection helpers.
+
+The app does not require Prometheus, Netdata, or sysstat. Instead it can install
+and manage a tiny POSIX-sh collector under /tmp on the remote server. The
+collector samples Linux /proc and /sys counters every 10 seconds and appends JSON
+Lines. This module starts/stops the collector and reads those JSONL samples.
+"""
+
+from __future__ import annotations
+
+import json
+import shlex
+from dataclasses import dataclass
+from typing import Any
+
+from media_library_viewer.clients.ssh import RemoteSSHClient
+
+# POSIX shell script copied to the remote server by start_resource_collector().
+# Keep this script bash-free because many NAS/media servers have minimal shells.
+COLLECTOR_SCRIPT = r'''#!/bin/sh
+set -u
+OUT="${1:-/tmp/media_library_viewer_metrics.jsonl}"
+INTERVAL="${2:-10}"
+RETENTION_SECONDS="${3:-604800}"
+MAX_LINES="${4:-70000}"
+PRUNE_EVERY_SAMPLES="${5:-60}"
+mkdir -p "$(dirname "$OUT")"
+
+echo "collector starting at $(date -Is 2>/dev/null || date), interval=${INTERVAL}s, retention=${RETENTION_SECONDS}s, max_lines=${MAX_LINES}, out=${OUT}"
+
+read_cpu() {
+  awk '/^cpu / {print $2+$3+$4+$5+$6+$7+$8+$9+$10, $5+$6}' /proc/stat
+}
+
+read_mem_pct() {
+  awk '
+    /^MemTotal:/ {total=$2}
+    /^MemAvailable:/ {avail=$2}
+    END {if (total > 0) printf "%.2f", (total-avail)*100/total; else printf "0"}
+  ' /proc/meminfo
+}
+
+read_net_bytes() {
+  awk '
+    NR > 2 {
+      split($0, parts, ":")
+      iface = parts[1]
+      stats = parts[2]
+      gsub(/^[ \t]+|[ \t]+$/, "", iface)
+      gsub(/^[ \t]+|[ \t]+$/, "", stats)
+      if (iface == "lo" || iface == "" || stats == "") next
+      split(stats, values, /[ \t]+/)
+      # /proc/net/dev after the colon:
+      # receive bytes are field 1, transmit bytes are field 9.
+      # Trim the stats block before split; otherwise leading whitespace can make
+      # values[1] empty in some awk implementations, resulting in zero rates.
+      rx += values[1] + 0
+      tx += values[9] + 0
+    }
+    END {printf "%.0f %.0f", rx, tx}
+  ' /proc/net/dev
+}
+
+read_disk_bytes() {
+  read_sectors=0
+  written_sectors=0
+  for dev in /sys/block/*; do
+    [ -r "$dev/stat" ] || continue
+    name="$(basename "$dev")"
+    case "$name" in
+      loop*|ram*|fd*|sr*) continue ;;
+    esac
+    # Linux /sys/block/<dev>/stat fields: 3=sectors read, 7=sectors written.
+    # Use POSIX sh parsing instead of bash arrays so this works on minimal systems.
+    set -- $(cat "$dev/stat")
+    sectors_read="${3:-0}"
+    sectors_written="${7:-0}"
+    read_sectors=$((read_sectors + sectors_read))
+    written_sectors=$((written_sectors + sectors_written))
+  done
+  printf "%s %s" "$((read_sectors * 512))" "$((written_sectors * 512))"
+}
+
+set -- $(read_cpu)
+prev_total="${1:-0}"
+prev_idle="${2:-0}"
+set -- $(read_net_bytes)
+prev_rx="${1:-0}"
+prev_tx="${2:-0}"
+set -- $(read_disk_bytes)
+prev_disk_read="${1:-0}"
+prev_disk_write="${2:-0}"
+prev_ts="$(date +%s)"
+sample_count=0
+
+prune_metrics_file() {
+  [ -f "$OUT" ] || return 0
+  cutoff="$1"
+  tmp="${OUT}.$$.tmp"
+  awk -v cutoff="$cutoff" '
+    match($0, /"ts":[0-9]+/) {
+      ts = substr($0, RSTART + 5, RLENGTH - 5)
+      if (ts >= cutoff) print $0
+    }
+  ' "$OUT" | tail -n "$MAX_LINES" > "$tmp" && mv "$tmp" "$OUT"
+  rm -f "$tmp"
+}
+
+while true; do
+  sleep "$INTERVAL"
+  now_ts="$(date +%s)"
+  dt=$((now_ts - prev_ts))
+  if [ "$dt" -le 0 ]; then dt=1; fi
+
+  set -- $(read_cpu)
+  total="${1:-0}"
+  idle="${2:-0}"
+  set -- $(read_net_bytes)
+  rx="${1:-0}"
+  tx="${2:-0}"
+  set -- $(read_disk_bytes)
+  disk_read="${1:-0}"
+  disk_write="${2:-0}"
+  mem_pct="$(read_mem_pct)"
+
+  total_delta=$((total - prev_total))
+  idle_delta=$((idle - prev_idle))
+  rx_delta=$((rx - prev_rx))
+  tx_delta=$((tx - prev_tx))
+  disk_read_delta=$((disk_read - prev_disk_read))
+  disk_write_delta=$((disk_write - prev_disk_write))
+
+  cpu_pct="$(awk -v total="$total_delta" -v idle="$idle_delta" 'BEGIN {if (total > 0) printf "%.2f", (total-idle)*100/total; else printf "0"}')"
+  rx_bytes_per_sec="$(awk -v bytes="$rx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
+  tx_bytes_per_sec="$(awk -v bytes="$tx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
+  rx_bps="$(awk -v bytes="$rx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes*8/dt; else printf "0"}')"
+  tx_bps="$(awk -v bytes="$tx_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes*8/dt; else printf "0"}')"
+  disk_read_bps="$(awk -v bytes="$disk_read_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
+  disk_write_bps="$(awk -v bytes="$disk_write_delta" -v dt="$dt" 'BEGIN {if (bytes >= 0) printf "%.0f", bytes/dt; else printf "0"}')"
+
+  printf '{"ts":%s,"cpu_pct":%s,"mem_pct":%s,"net_rx_bytes_per_sec":%s,"net_tx_bytes_per_sec":%s,"net_rx_bps":%s,"net_tx_bps":%s,"disk_read_bps":%s,"disk_write_bps":%s}\n' \
+    "$now_ts" "$cpu_pct" "$mem_pct" "$rx_bytes_per_sec" "$tx_bytes_per_sec" "$rx_bps" "$tx_bps" "$disk_read_bps" "$disk_write_bps" >> "$OUT"
+
+  sample_count=$((sample_count + 1))
+  if [ $((sample_count % PRUNE_EVERY_SAMPLES)) -eq 0 ]; then
+    prune_metrics_file "$((now_ts - RETENTION_SECONDS))"
+  fi
+
+  prev_total="$total"
+  prev_idle="$idle"
+  prev_rx="$rx"
+  prev_tx="$tx"
+  prev_disk_read="$disk_read"
+  prev_disk_write="$disk_write"
+  prev_ts="$now_ts"
+done
+'''
+
+
+@dataclass(frozen=True)
+class ResourceMonitorPaths:
+    """Remote file locations used by the lightweight resource collector."""
+
+    metrics_file: str = "/tmp/media_library_viewer_metrics.jsonl"
+    pid_file: str = "/tmp/media_library_viewer_metrics.pid"
+    script_file: str = "/tmp/media_library_viewer_metrics_collector.sh"
+    log_file: str = "/tmp/media_library_viewer_metrics.log"
+
+
+def start_resource_collector(
+    ssh: RemoteSSHClient,
+    interval_seconds: int = 10,
+    retention_seconds: int = 7 * 24 * 60 * 60,
+    max_lines: int = 70_000,
+    paths: ResourceMonitorPaths = ResourceMonitorPaths(),
+) -> str:
+    """Install and start the remote metrics collector if it is not running.
+
+    Starting a fresh collector removes old metrics/log files because schema
+    changes during development can otherwise leave mixed JSONL records behind.
+    The collector prunes its own metrics file to 7 days / max_lines.
+    """
+    command = f"""
+cat > {shlex.quote(paths.script_file)} <<'MLV_RESOURCE_COLLECTOR'
+{COLLECTOR_SCRIPT}
+MLV_RESOURCE_COLLECTOR
+chmod +x {shlex.quote(paths.script_file)}
+if [ -f {shlex.quote(paths.pid_file)} ] && kill -0 "$(cat {shlex.quote(paths.pid_file)})" 2>/dev/null; then
+  echo "already running pid=$(cat {shlex.quote(paths.pid_file)})"
+else
+  rm -f {shlex.quote(paths.metrics_file)} {shlex.quote(paths.log_file)}
+  nohup {shlex.quote(paths.script_file)} {shlex.quote(paths.metrics_file)} {int(interval_seconds)} {int(retention_seconds)} {int(max_lines)} >> {shlex.quote(paths.log_file)} 2>&1 &
+  echo $! > {shlex.quote(paths.pid_file)}
+  echo "started pid=$(cat {shlex.quote(paths.pid_file)})"
+fi
+"""
+    result = ssh.run(command, timeout=20)
+    if result.exit_status != 0:
+        raise RuntimeError(result.stderr or result.stdout or "failed to start resource collector")
+    return result.stdout.strip()
+
+
+def stop_resource_collector(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
+    """Stop the remote collector process if the pid file points to one."""
+    command = f"""
+if [ -f {shlex.quote(paths.pid_file)} ]; then
+  pid="$(cat {shlex.quote(paths.pid_file)})"
+  if kill -0 "$pid" 2>/dev/null; then
+    kill "$pid"
+    echo "stopped pid=$pid"
+  else
+    echo "not running"
+  fi
+  rm -f {shlex.quote(paths.pid_file)}
+else
+  echo "not running"
+fi
+"""
+    result = ssh.run(command, timeout=20)
+    if result.exit_status != 0:
+        raise RuntimeError(result.stderr or result.stdout or "failed to stop resource collector")
+    return result.stdout.strip()
+
+
+def restart_resource_collector(
+    ssh: RemoteSSHClient,
+    interval_seconds: int = 10,
+    retention_seconds: int = 7 * 24 * 60 * 60,
+    max_lines: int = 70_000,
+    paths: ResourceMonitorPaths = ResourceMonitorPaths(),
+) -> str:
+    stop_message = stop_resource_collector(ssh, paths)
+    start_message = start_resource_collector(ssh, interval_seconds, retention_seconds, max_lines, paths)
+    return f"{stop_message}\n{start_message}"
+
+
+def resource_collector_status(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
+    """Return a short human-readable status string for the dashboard."""
+    command = f"""
+if [ -f {shlex.quote(paths.pid_file)} ] && kill -0 "$(cat {shlex.quote(paths.pid_file)})" 2>/dev/null; then
+  echo "running pid=$(cat {shlex.quote(paths.pid_file)})"
+else
+  echo "not running"
+fi
+"""
+    result = ssh.run(command, timeout=10)
+    if result.exit_status != 0:
+        raise RuntimeError(result.stderr or result.stdout or "failed to check collector status")
+    return result.stdout.strip()
+
+
+def resource_collector_debug_info(ssh: RemoteSSHClient, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> str:
+    """Collect remote diagnostics for troubleshooting missing metrics."""
+    command = f"""
+echo "status:"
+if [ -f {shlex.quote(paths.pid_file)} ]; then
+  pid="$(cat {shlex.quote(paths.pid_file)})"
+  echo "pid_file=$pid"
+  if kill -0 "$pid" 2>/dev/null; then echo "process=running"; else echo "process=not-running"; fi
+else
+  echo "pid_file=missing"
+fi
+
+echo "files:"
+ls -l {shlex.quote(paths.script_file)} {shlex.quote(paths.metrics_file)} {shlex.quote(paths.log_file)} 2>&1 || true
+
+echo "sample_count:"
+if [ -f {shlex.quote(paths.metrics_file)} ]; then wc -l < {shlex.quote(paths.metrics_file)}; else echo 0; fi
+
+echo "last_samples:"
+if [ -f {shlex.quote(paths.metrics_file)} ]; then tail -n 5 {shlex.quote(paths.metrics_file)}; fi
+
+echo "log_tail:"
+if [ -f {shlex.quote(paths.log_file)} ]; then tail -n 40 {shlex.quote(paths.log_file)}; fi
+
+echo "netdev_snapshot:"
+cat /proc/net/dev 2>&1 || true
+"""
+    result = ssh.run(command, timeout=20)
+    return (result.stdout or "") + (result.stderr or "")
+
+
+def read_resource_metrics(ssh: RemoteSSHClient, max_lines: int = 1000, paths: ResourceMonitorPaths = ResourceMonitorPaths()) -> list[dict[str, Any]]:
+    """Read recent JSONL metric samples from the remote collector file."""
+    command = f"test -f {shlex.quote(paths.metrics_file)} && tail -n {int(max_lines)} {shlex.quote(paths.metrics_file)} || true"
+    result = ssh.run(command, timeout=20)
+    if result.exit_status != 0:
+        raise RuntimeError(result.stderr or result.stdout or "failed to read resource metrics")
+    rows = []
+    for line in result.stdout.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            rows.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return rows
+
+
+def disk_space(ssh: RemoteSSHClient, path: str = "/") -> dict[str, Any]:
+    """Return df information for the filesystem containing ``path``."""
+    command = (
+        "df -P -B1 -- "
+        + shlex.quote(path or "/")
+        + " | awk 'NR==2 {printf \"{\\\"filesystem\\\":\\\"%s\\\",\\\"size\\\":%s,\\\"used\\\":%s,\\\"available\\\":%s,\\\"used_pct\\\":\\\"%s\\\",\\\"mount\\\":\\\"%s\\\"}\", $1,$2,$3,$4,$5,$6}'"
+    )
+    result = ssh.run(command, timeout=20)
+    if result.exit_status != 0 or not result.stdout.strip():
+        raise RuntimeError(result.stderr or result.stdout or "failed to read disk space")
+    return json.loads(result.stdout)
diff --git a/src/media_library_viewer/clients/ssh.py b/src/media_library_viewer/clients/ssh.py
new file mode 100644
index 0000000..0443b14
--- /dev/null
+++ b/src/media_library_viewer/clients/ssh.py
@@ -0,0 +1,146 @@
+"""SSH client helpers for remote filesystem and media inspection.
+
+All command execution goes through ``/bin/sh -c`` and all paths inserted into
+commands are shell-quoted by callers. This is important for two reasons:
+
+1. The remote login shell may be fish/csh/etc.; internal commands are POSIX sh.
+2. Media paths frequently contain spaces and punctuation.
+"""
+
+from __future__ import annotations
+
+import json
+import posixpath
+import shlex
+from dataclasses import dataclass
+from typing import Any
+
+import paramiko
+
+
+@dataclass
+class CommandResult:
+    """Plain result object returned by remote command execution."""
+
+    command: str
+    exit_status: int
+    stdout: str
+    stderr: str
+
+
+class RemoteSSHClient:
+    """SSH helper for read-only inspection plus explicit job execution."""
+
+    def __init__(
+        self,
+        host: str,
+        username: str,
+        port: int = 22,
+        key_filename: str | None = None,
+        password: str | None = None,
+        timeout: int = 20,
+    ):
+        if not host or not username:
+            raise ValueError("SSH host and username are required")
+        self.host = host
+        self.username = username
+        self.port = port
+        self.key_filename = key_filename or None
+        self.password = password or None
+        self.timeout = timeout
+        self._client: paramiko.SSHClient | None = None
+
+    def connect(self) -> paramiko.SSHClient:
+        """Create or reuse the Paramiko connection.
+
+        Unknown host keys are rejected. Users should connect once manually with
+        ssh so the server is present in known_hosts.
+        """
+        if self._client:
+            return self._client
+        client = paramiko.SSHClient()
+        client.load_system_host_keys()
+        client.set_missing_host_key_policy(paramiko.RejectPolicy())
+        client.connect(
+            self.host,
+            port=self.port,
+            username=self.username,
+            key_filename=self.key_filename,
+            password=self.password,
+            timeout=self.timeout,
+        )
+        self._client = client
+        return client
+
+    def close(self) -> None:
+        if self._client:
+            self._client.close()
+            self._client = None
+
+    def run(self, command: str, timeout: int | None = None) -> CommandResult:
+        """Run a command through POSIX sh, independent of the user's login shell.
+
+        Paramiko asks the SSH server to execute a command using the account's
+        default shell. If that shell is fish/csh/etc., POSIX snippets containing
+        `if ...; then`, pipes, redirects, or heredocs can fail. All internal app
+        commands and job templates are written for POSIX shell, so explicitly
+        dispatch through `/bin/sh -c`.
+        """
+        client = self.connect()
+        shell_command = f"/bin/sh -c {shlex.quote(command)}"
+        stdin, stdout, stderr = client.exec_command(shell_command, timeout=timeout or self.timeout)
+        exit_status = stdout.channel.recv_exit_status()
+        return CommandResult(
+            command=command,
+            exit_status=exit_status,
+            stdout=stdout.read().decode(errors="replace"),
+            stderr=stderr.read().decode(errors="replace"),
+        )
+
+    def list_dir(self, path: str) -> CommandResult:
+        """List one remote directory as JSON.
+
+        The command first verifies that ``path`` is a directory. Without that
+        guard, running ``find`` on a file can look like an empty directory, which
+        was a source of file-browser confusion. Output is NUL-delimited before
+        Python serializes it, making spaces in filenames safe.
+        """
+        # JSON-ish output: type, size, mtime epoch, filename. Handles spaces/newlines reasonably via NUL boundaries.
+        quoted = shlex.quote(path)
+        not_dir_message = shlex.quote(f"Not a directory: {path}")
+        command = (
+            f"test -d {quoted} || "
+            f"{{ echo {not_dir_message} >&2; exit 20; }}; "
+            f"find {quoted} -maxdepth 1 -mindepth 1 -printf "
+            "'%y\\t%s\\t%T@\\t%f\\0' | python3 -c "
+            + shlex.quote(
+                "import sys,json; data=sys.stdin.buffer.read().split(b'\\0'); "
+                "rows=[]\n"
+                "for row in data:\n"
+                "    if not row: continue\n"
+                "    t,s,m,n=row.decode('utf-8','replace').split('\\t',3)\n"
+                "    rows.append({'type':t,'size':int(s),'mtime':float(m),'name':n})\n"
+                "print(json.dumps(rows))"
+            )
+        )
+        return self.run(command)
+
+    def stat_path(self, path: str) -> CommandResult:
+        """Run stat for a remote file or directory path."""
+        quoted = shlex.quote(path)
+        return self.run(f"stat --printf='%F\\n%s bytes\\n%y\\n%n\\n' {quoted}")
+
+    def ffprobe_json(self, path: str) -> dict[str, Any]:
+        """Run ffprobe and parse JSON output for a remote media file."""
+        quoted = shlex.quote(path)
+        result = self.run(
+            "ffprobe -v error -show_format -show_streams -print_format json " + quoted,
+            timeout=60,
+        )
+        if result.exit_status != 0:
+            raise RuntimeError(result.stderr or result.stdout or "ffprobe failed")
+        return json.loads(result.stdout)
+
+    @staticmethod
+    def join(parent: str, child: str) -> str:
+        return posixpath.normpath(posixpath.join(parent, child))
diff --git a/src/media_library_viewer/config.py b/src/media_library_viewer/config.py
new file mode 100644
index 0000000..e922f49
--- /dev/null
+++ b/src/media_library_viewer/config.py
@@ -0,0 +1,49 @@
+"""Configuration loading for the app.
+
+Configuration is intentionally environment/.env based so credentials stay out of
+source control and the same package can be reused by different frontends or
+process managers.
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+@dataclass(frozen=True)
+class JellyfinConfig:
+    """Jellyfin connection settings."""
+
+    url: str = os.getenv("JELLYFIN_URL", "")
+    api_key: str = os.getenv("JELLYFIN_API_KEY", "")
+    user_id: str = os.getenv("JELLYFIN_USER_ID", "")
+
+
+@dataclass(frozen=True)
+class SSHConfig:
+    """SSH connection settings for remote file/resource access."""
+
+    host: str = os.getenv("SSH_HOST", "")
+    username: str = os.getenv("SSH_USERNAME", "")
+    port: int = int(os.getenv("SSH_PORT", "22"))
+    key_filename: str = os.getenv("SSH_KEY_FILENAME", str(Path.home() / ".ssh" / "id_rsa"))
+    password: str = os.getenv("SSH_PASSWORD", "")
+    media_root: str = os.getenv("REMOTE_MEDIA_ROOT", "")
+    path_prefix: str = os.getenv("REMOTE_PATH_PREFIX", "")
+
+
+@dataclass(frozen=True)
+class AppConfig:
+    jellyfin: JellyfinConfig = JellyfinConfig()
+    ssh: SSHConfig = SSHConfig()
+
+
+def load_config() -> AppConfig:
+    """Build an AppConfig snapshot from the current environment/.env file."""
+    return AppConfig()
diff --git a/src/media_library_viewer/domain/__init__.py b/src/media_library_viewer/domain/__init__.py
new file mode 100644
index 0000000..98ba851
--- /dev/null
+++ b/src/media_library_viewer/domain/__init__.py
@@ -0,0 +1,5 @@
+"""Domain-level helpers and normalization code.
+
+Domain modules convert external data into stable app concepts and should remain
+independent of Streamlit or any future frontend framework.
+"""
diff --git a/src/media_library_viewer/domain/media.py b/src/media_library_viewer/domain/media.py
new file mode 100644
index 0000000..84e0e2c
--- /dev/null
+++ b/src/media_library_viewer/domain/media.py
@@ -0,0 +1,161 @@
+"""Media-domain normalization helpers.
+
+Jellyfin item JSON is nested and inconsistent across item types. This module
+flattens Jellyfin items into stable dictionaries suitable for storage in the
+SQLite media index and display by any frontend.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+
+from media_library_viewer.utils import human_size, ticks_to_minutes
+
+
+def first_media_source(item: dict[str, Any]) -> dict[str, Any]:
+    """Return the first Jellyfin media source, or an empty dict."""
+    sources = item.get("MediaSources") or []
+    return sources[0] if sources else {}
+
+
+def media_streams(item: dict[str, Any], stream_type: str | None = None) -> list[dict[str, Any]]:
+    """Return flattened media streams from all media sources.
+
+    Jellyfin usually nests streams under MediaSources, while some endpoints may
+    expose stream-like fields differently. This function gives callers one place
+    to get streams and optionally filter by type.
+    """
+    streams = []
+    for source in item.get("MediaSources") or []:
+        streams.extend(source.get("MediaStreams") or [])
+    if stream_type is None:
+        return streams
+    return [stream for stream in streams if str(stream.get("Type") or stream.get("codec_type") or "").lower() == stream_type.lower()]
+
+
+def stream_value(stream: dict[str, Any], *keys: str) -> Any:
+    for key in keys:
+        if key in stream and stream[key] not in (None, ""):
+            return stream[key]
+    return None
+
+
+def is_hdr_item(item: dict[str, Any]) -> bool:
+    """Best-effort HDR detection from Jellyfin video stream metadata."""
+    hdr_markers = {"hdr", "hdr10", "hdr10+", "dolbyvision", "dovi", "hlg", "pq", "smpte2084", "bt2020"}
+    for stream in media_streams(item, "Video"):
+        values = [
+            stream_value(stream, "VideoRange", "video_range"),
+            stream_value(stream, "VideoRangeType", "video_range_type"),
+            stream_value(stream, "ColorTransfer", "color_transfer"),
+            stream_value(stream, "ColorPrimaries", "color_primaries"),
+            stream_value(stream, "ColorSpace", "color_space"),
+            stream_value(stream, "DvVersionMajor", "dv_version_major"),
+            stream_value(stream, "Hdr10PlusPresent", "hdr10_plus_present"),
+        ]
+        normalized = " ".join(str(value).lower() for value in values if value not in (None, "", False, 0))
+        if any(marker in normalized for marker in hdr_markers):
+            return True
+    return False
+
+
+def format_date_added(value: str | None) -> str:
+    if not value:
+        return ""
+    try:
+        return pd.to_datetime(value).strftime("%Y-%m-%d")
+    except Exception:
+        return str(value)
+
+
+def timestamp_date_added(value: str | None) -> int | None:
+    if not value:
+        return None
+    try:
+        return int(pd.to_datetime(value).timestamp())
+    except Exception:
+        return None
+
+
+def format_rate_bits_decimal(bits_per_second: float | int | str | None) -> str:
+    if bits_per_second in (None, ""):
+        return ""
+    try:
+        value = float(bits_per_second)
+    except (TypeError, ValueError):
+        return str(bits_per_second)
+    for unit in ["bps", "Kbps", "Mbps", "Gbps", "Tbps"]:
+        if value < 1000 or unit == "Tbps":
+            return f"{value:.1f} {unit}"
+        value /= 1000
+    return f"{value:.1f} Tbps"
+
+
+def normalize_media_item(item: dict[str, Any], library_id: str = "", library_name: str = "") -> dict[str, Any]:
+    """Flatten one Jellyfin item into an indexable row.
+
+    The returned row contains both display strings (``size``, ``bitrate``) and
+    numeric sort fields (``size_bytes``, ``bitrate_bps``, ``date_added_ts``).
+    """
+    source = first_media_source(item)
+    video_streams = media_streams(item, "Video")
+    video = video_streams[0] if video_streams else {}
+    size = source.get("Size") or source.get("size")
+    bitrate = source.get("Bitrate") or source.get("bitrate") or item.get("Bitrate")
+    width = stream_value(video, "Width", "width")
+    height = stream_value(video, "Height", "height")
+    season_number = item.get("ParentIndexNumber")
+    episode_number = item.get("IndexNumber")
+
+    hdr = is_hdr_item(item)
+    return {
+        "id": item.get("Id", ""),
+        "title": item.get("Name", ""),
+        "series": item.get("SeriesName", ""),
+        "season": f"S{int(season_number):02d}" if season_number is not None else item.get("SeasonName", ""),
+        "season_number": int(season_number) if season_number is not None else None,
+        "episode": int(episode_number) if episode_number is not None else None,
+        "type": item.get("Type", ""),
+        "year": item.get("ProductionYear"),
+        "runtime_ticks": item.get("RunTimeTicks"),
+        "runtime_min": ticks_to_minutes(item.get("RunTimeTicks")),
+        "size_bytes": int(size) if size not in (None, "") else None,
+        "size": human_size(size),
+        "bitrate_bps": int(bitrate) if bitrate not in (None, "") else None,
+        "bitrate": format_rate_bits_decimal(bitrate),
+        "hdr": 1 if hdr else 0,
+        "hdr_label": "yes" if hdr else "",
+        "video": video.get("Codec") or video.get("codec_name") or "",
+        "width": int(width) if width not in (None, "") else None,
+        "height": int(height) if height not in (None, "") else None,
+        "resolution": f"{width}x{height}" if width and height else "",
+        "date_added": format_date_added(item.get("DateCreated")),
+        "date_added_ts": timestamp_date_added(item.get("DateCreated")),
+        "path": item.get("Path") or source.get("Path") or "",
+        "library_id": library_id,
+        "library_name": library_name,
+    }
+
+
+def display_media_row(row: dict[str, Any]) -> dict[str, Any]:
+    """Convert a SQLite row back into frontend display fields."""
+    return {
+        "title": row.get("title", ""),
+        "series": row.get("series", ""),
+        "season": row.get("season", ""),
+        "episode": row.get("episode", ""),
+        "type": row.get("type", ""),
+        "year": row.get("year", ""),
+        "runtime_min": row.get("runtime_min", ""),
+        "size": row.get("size") or human_size(row.get("size_bytes")),
+        "bitrate": row.get("bitrate") or format_rate_bits_decimal(row.get("bitrate_bps")),
+        "hdr": "yes" if row.get("hdr") else "no",
+        "video": row.get("video", ""),
+        "resolution": row.get("resolution", ""),
+        "date_added": row.get("date_added", ""),
+        "library": row.get("library_name", ""),
+        "path": row.get("path", ""),
+        "id": row.get("id", ""),
+    }
diff --git a/src/media_library_viewer/jobs.py b/src/media_library_viewer/jobs.py
new file mode 100644
index 0000000..682ebb2
--- /dev/null
+++ b/src/media_library_viewer/jobs.py
@@ -0,0 +1,59 @@
+"""Template-based remote jobs.
+
+Remote jobs are intentionally explicit templates instead of free-form shell input.
+This keeps the UI safer and makes future destructive operations easier to wrap in
+confirmations/dry-runs.
+"""
+
+from __future__ import annotations
+
+import shlex
+from dataclasses import dataclass
+from typing import Mapping
+
+from media_library_viewer.clients.ssh import CommandResult, RemoteSSHClient
+
+
+@dataclass(frozen=True)
+class JobTemplate:
+    """Description and command template for one remote job."""
+
+    name: str
+    description: str
+    command_template: str
+    destructive: bool = False
+
+    def render(self, values: Mapping[str, str]) -> str:
+        """Render the command with shell-quoted template values.
+
+        This is what keeps paths with spaces safe when inserted into job commands.
+        """
+        safe_values = {key: shlex.quote(value) for key, value in values.items()}
+        return self.command_template.format(**safe_values)
+
+
+# Phase 1 jobs are intentionally conservative. Add your own templates here later.
+JOB_TEMPLATES: dict[str, JobTemplate] = {
+    "disk_usage": JobTemplate(
+        name="Disk usage for selected path",
+        description="Runs du -sh on the selected remote path.",
+        command_template="du -sh {path}",
+    ),
+    "ffprobe": JobTemplate(
+        name="ffprobe JSON",
+        description="Prints raw ffprobe stream/format metadata.",
+        command_template="ffprobe -v error -show_format -show_streams -print_format json {path}",
+    ),
+    "dry_run_find_empty_dirs": JobTemplate(
+        name="Find empty directories dry-run",
+        description="Lists empty directories under the selected path. Does not delete anything.",
+        command_template="find {path} -type d -empty -print",
+    ),
+}
+
+
+def run_job(ssh: RemoteSSHClient, job_key: str, path: str, timeout: int = 600) -> CommandResult:
+    """Render and execute a configured job template for a selected remote path."""
+    template = JOB_TEMPLATES[job_key]
+    command = template.render({"path": path})
+    return ssh.run(command, timeout=timeout)
diff --git a/src/media_library_viewer/services/__init__.py b/src/media_library_viewer/services/__init__.py
new file mode 100644
index 0000000..68c0f05
--- /dev/null
+++ b/src/media_library_viewer/services/__init__.py
@@ -0,0 +1,5 @@
+"""Application services.
+
+Services coordinate clients/domain logic into reusable operations. They are the
+natural layer to expose through a future HTTP API for a React frontend.
+"""
diff --git a/src/media_library_viewer/services/media_index.py b/src/media_library_viewer/services/media_index.py
new file mode 100644
index 0000000..9c70acd
--- /dev/null
+++ b/src/media_library_viewer/services/media_index.py
@@ -0,0 +1,278 @@
+"""SQLite-backed media inventory service.
+
+This is the main step toward a frontend-agnostic architecture. The Streamlit UI
+asks this service to build/query an index, but the same class could be exposed
+through FastAPI to a React frontend without rewriting Jellyfin indexing logic.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+from media_library_viewer.clients.jellyfin import JellyfinClient
+from media_library_viewer.domain.media import display_media_row, normalize_media_item
+
+# Local generated database. It is ignored by git and can be rebuilt from
+# Jellyfin metadata whenever needed.
+DEFAULT_INDEX_PATH = Path(".cache/media_library_viewer/media_index.sqlite")
+MEDIA_TYPES = "Movie,Episode,Video"
+
+# Only values from this whitelist are interpolated into ORDER BY. User-selected
+# sort keys map to these known SQL snippets to avoid SQL injection.
+SORT_COLUMNS = {
+    "title": "title COLLATE NOCASE",
+    "series": "series COLLATE NOCASE",
+    "season": "season_number",
+    "episode": "episode",
+    "type": "type COLLATE NOCASE",
+    "year": "year",
+    "runtime": "runtime_min",
+    "size": "size_bytes",
+    "bitrate": "bitrate_bps",
+    "hdr": "hdr",
+    "video": "video COLLATE NOCASE",
+    "resolution": "height",
+    "date_added": "date_added_ts",
+    "library": "library_name COLLATE NOCASE",
+    "path": "path COLLATE NOCASE",
+}
+
+
+@dataclass(frozen=True)
+class MediaIndexStatus:
+    """Lightweight status object displayed by the Media tab."""
+
+    exists: bool
+    item_count: int = 0
+    updated_at: int | None = None
+    updated_at_label: str = ""
+    build_duration_seconds: float | None = None
+
+
+class MediaIndex:
+    """SQLite-backed media inventory.
+
+    This class is UI-framework independent. Streamlit, a future FastAPI backend,
+    or a React-facing API can all use this service.
+    """
+
+    def __init__(self, db_path: Path | str = DEFAULT_INDEX_PATH):
+        self.db_path = Path(db_path)
+        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+
+    def connect(self) -> sqlite3.Connection:
+        """Open a sqlite connection configured to return Row objects."""
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def init_schema(self) -> None:
+        """Create tables/indexes if this is the first use of the index."""
+        with self.connect() as conn:
+            conn.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS media_items (
+                    id TEXT PRIMARY KEY,
+                    title TEXT,
+                    series TEXT,
+                    season TEXT,
+                    season_number INTEGER,
+                    episode INTEGER,
+                    type TEXT,
+                    year INTEGER,
+                    runtime_ticks INTEGER,
+                    runtime_min INTEGER,
+                    size_bytes INTEGER,
+                    bitrate_bps INTEGER,
+                    hdr INTEGER,
+                    video TEXT,
+                    width INTEGER,
+                    height INTEGER,
+                    resolution TEXT,
+                    date_added TEXT,
+                    date_added_ts INTEGER,
+                    path TEXT,
+                    library_id TEXT,
+                    library_name TEXT
+                );
+                CREATE TABLE IF NOT EXISTS index_metadata (
+                    key TEXT PRIMARY KEY,
+                    value TEXT
+                );
+                CREATE INDEX IF NOT EXISTS idx_media_type ON media_items(type);
+                CREATE INDEX IF NOT EXISTS idx_media_library ON media_items(library_id);
+                CREATE INDEX IF NOT EXISTS idx_media_title ON media_items(title COLLATE NOCASE);
+                CREATE INDEX IF NOT EXISTS idx_media_series ON media_items(series COLLATE NOCASE);
+                CREATE INDEX IF NOT EXISTS idx_media_date_added ON media_items(date_added_ts);
+                CREATE INDEX IF NOT EXISTS idx_media_size ON media_items(size_bytes);
+                CREATE INDEX IF NOT EXISTS idx_media_bitrate ON media_items(bitrate_bps);
+                """
+            )
+
+    def set_metadata(self, key: str, value: str | int | float) -> None:
+        """Store a small string metadata value, e.g. build duration."""
+        self.init_schema()
+        with self.connect() as conn:
+            conn.execute(
+                "INSERT OR REPLACE INTO index_metadata (key, value) VALUES (?, ?)",
+                (key, str(value)),
+            )
+
+    def replace_items(self, rows: Iterable[dict[str, Any]]) -> int:
+        """Atomically replace indexed media rows with a freshly built set."""
+        self.init_schema()
+        row_list = list(rows)
+        columns = [
+            "id",
+            "title",
+            "series",
+            "season",
+            "season_number",
+            "episode",
+            "type",
+            "year",
+            "runtime_ticks",
+            "runtime_min",
+            "size_bytes",
+            "bitrate_bps",
+            "hdr",
+            "video",
+            "width",
+            "height",
+            "resolution",
+            "date_added",
+            "date_added_ts",
+            "path",
+            "library_id",
+            "library_name",
+        ]
+        placeholders = ",".join(["?"] * len(columns))
+        with self.connect() as conn:
+            conn.execute("DELETE FROM media_items")
+            conn.executemany(
+                f"INSERT OR REPLACE INTO media_items ({','.join(columns)}) VALUES ({placeholders})",
+                [[row.get(column) for column in columns] for row in row_list],
+            )
+            conn.execute(
+                "INSERT OR REPLACE INTO index_metadata (key, value) VALUES ('updated_at', ?)",
+                (str(int(time.time())),),
+            )
+        return len(row_list)
+
+    def status(self) -> MediaIndexStatus:
+        """Return existence, count, update time, and last build duration."""
+        if not self.db_path.exists():
+            return MediaIndexStatus(exists=False)
+        try:
+            with self.connect() as conn:
+                item_count = int(conn.execute("SELECT COUNT(*) FROM media_items").fetchone()[0])
+                updated_row = conn.execute("SELECT value FROM index_metadata WHERE key='updated_at'").fetchone()
+                duration_row = conn.execute("SELECT value FROM index_metadata WHERE key='build_duration_seconds'").fetchone()
+        except sqlite3.Error:
+            return MediaIndexStatus(exists=False)
+        updated_at = int(updated_row[0]) if updated_row and str(updated_row[0]).isdigit() else None
+        label = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(updated_at)) if updated_at else ""
+        build_duration = None
+        if duration_row:
+            try:
+                build_duration = float(duration_row[0])
+            except (TypeError, ValueError):
+                build_duration = None
+        return MediaIndexStatus(
+            exists=True,
+            item_count=item_count,
+            updated_at=updated_at,
+            updated_at_label=label,
+            build_duration_seconds=build_duration,
+        )
+
+    def query(
+        self,
+        library_id: str | None = None,
+        library_ids: list[str] | None = None,
+        media_types: list[str] | None = None,
+        search: str = "",
+        hdr_filter: str = "All",
+        sort_key: str = "title",
+        sort_order: str = "Ascending",
+        limit: int = 100,
+        offset: int = 0,
+    ) -> tuple[list[dict[str, Any]], int]:
+        """Query indexed media with full-index filters, sorting, and pagination."""
+        self.init_schema()
+        where = []
+        params: list[Any] = []
+        if library_ids:
+            where.append("library_id IN (" + ",".join(["?"] * len(library_ids)) + ")")
+            params.extend(library_ids)
+        elif library_id:
+            where.append("library_id = ?")
+            params.append(library_id)
+        if media_types:
+            where.append("type IN (" + ",".join(["?"] * len(media_types)) + ")")
+            params.extend(media_types)
+        if search:
+            needle = f"%{search.lower()}%"
+            where.append("(LOWER(title) LIKE ? OR LOWER(series) LIKE ? OR LOWER(path) LIKE ?)")
+            params.extend([needle, needle, needle])
+        if hdr_filter == "HDR only":
+            where.append("hdr = 1")
+        elif hdr_filter == "SDR/unknown only":
+            where.append("(hdr IS NULL OR hdr = 0)")
+
+        where_sql = " WHERE " + " AND ".join(where) if where else ""
+        sort_sql = SORT_COLUMNS.get(sort_key, SORT_COLUMNS["title"])
+        direction = "DESC" if sort_order == "Descending" else "ASC"
+        # Always add stable tie-breakers.
+        order_sql = f" ORDER BY {sort_sql} {direction}, series COLLATE NOCASE ASC, season_number ASC, episode ASC, title COLLATE NOCASE ASC"
+
+        with self.connect() as conn:
+            total = int(conn.execute("SELECT COUNT(*) FROM media_items" + where_sql, params).fetchone()[0])
+            rows = conn.execute(
+                "SELECT * FROM media_items" + where_sql + order_sql + " LIMIT ? OFFSET ?",
+                [*params, int(limit), int(offset)],
+            ).fetchall()
+        return [display_media_row(dict(row)) for row in rows], total
+
+
+def build_media_index(
+    client: JellyfinClient,
+    user_id: str,
+    libraries: list[dict[str, Any]],
+    index: MediaIndex | None = None,
+    page_size: int = 500,
+) -> int:
+    """Fetch Jellyfin pages for all selected libraries and rebuild the index."""
+    index = index or MediaIndex()
+    started_at = time.perf_counter()
+    normalized_rows: list[dict[str, Any]] = []
+    for library in libraries:
+        library_id = library.get("Id")
+        library_name = library.get("Name", "")
+        if not library_id:
+            continue
+        start = 0
+        while True:
+            response = client.items(
+                user_id=user_id,
+                parent_id=library_id,
+                start_index=start,
+                limit=page_size,
+                include_item_types=MEDIA_TYPES,
+                recursive=True,
+                sort_by="SortName",
+                sort_order="Ascending",
+            )
+            items = response.get("Items", [])
+            normalized_rows.extend(normalize_media_item(item, library_id, library_name) for item in items)
+            start += len(items)
+            total = int(response.get("TotalRecordCount", start))
+            if not items or start >= total:
+                break
+    count = index.replace_items(normalized_rows)
+    index.set_metadata("build_duration_seconds", f"{time.perf_counter() - started_at:.3f}")
+    return count
diff --git a/src/media_library_viewer/ui/__init__.py b/src/media_library_viewer/ui/__init__.py
new file mode 100644
index 0000000..80ea336
--- /dev/null
+++ b/src/media_library_viewer/ui/__init__.py
@@ -0,0 +1,5 @@
+"""Streamlit UI modules.
+
+Each module renders one major slice of the application so the main app entrypoint
+stays small and future frontend replacement is easier to reason about.
+"""
diff --git a/src/media_library_viewer/ui/dashboard.py b/src/media_library_viewer/ui/dashboard.py
new file mode 100644
index 0000000..5c4da9d
--- /dev/null
+++ b/src/media_library_viewer/ui/dashboard.py
@@ -0,0 +1,276 @@
+"""Dashboard and resource-tab UI."""
+
+from __future__ import annotations
+
+import time
+from typing import Any
+
+import pandas as pd
+import streamlit as st
+
+from media_library_viewer.clients.resources import (
+    disk_space,
+    read_resource_metrics,
+    resource_collector_debug_info,
+    resource_collector_status,
+    restart_resource_collector,
+    start_resource_collector,
+    stop_resource_collector,
+)
+from media_library_viewer.utils import human_size
+
+
+def format_rate_bytes(bytes_per_second: float | int | None) -> str:
+    if bytes_per_second is None:
+        return ""
+    return f"{human_size(bytes_per_second)}/s"
+
+
+def rate_scale(max_value: float | int | None) -> tuple[float, str]:
+    value = abs(float(max_value or 0))
+    units = [(1_000_000_000_000, "TB/s"), (1_000_000_000, "GB/s"), (1_000_000, "MB/s"), (1_000, "KB/s"), (1, "B/s")]
+    for divisor, suffix in units:
+        if value >= divisor or divisor == 1:
+            return float(divisor), suffix
+    return 1.0, units[-1][1]
+
+
+def scaled_rate_chart_df(chart_df: pd.DataFrame, columns: list[str], labels: list[str]) -> tuple[pd.DataFrame, str]:
+    max_value = chart_df[columns].max(numeric_only=True).max()
+    divisor, suffix = rate_scale(max_value)
+    scaled = chart_df[columns].copy() / divisor
+    scaled.columns = [f"{label} ({suffix})" for label in labels]
+    return scaled, suffix
+
+
+def format_elapsed(seconds: float | int | None) -> str:
+    if seconds is None:
+        return ""
+    seconds = float(seconds)
+    if seconds < 60:
+        return f"{seconds:.1f}s"
+    minutes = int(seconds // 60)
+    remainder = seconds % 60
+    if minutes < 60:
+        return f"{minutes}m {remainder:.0f}s"
+    hours = minutes // 60
+    minutes = minutes % 60
+    return f"{hours}h {minutes}m"
+
+
+def render_media_overview(cached_media_counts, base_url: str, api_key: str, user_id: str) -> None:
+    """Render dashboard counts for movies/series/episodes."""
+    st.subheader("Media library overview")
+    try:
+        counts = cached_media_counts(base_url, api_key, user_id)
+    except Exception as exc:
+        st.warning(f"Could not load Jellyfin media counts: {exc}")
+        return
+
+    cols = st.columns(3)
+    cols[0].metric("Movies", f"{counts.get('movies', 0):,}")
+    cols[1].metric("Series", f"{counts.get('series', 0):,}")
+    cols[2].metric("Episodes", f"{counts.get('episodes', 0):,}")
+
+
+def render_now_playing(cached_active_sessions, base_url: str, api_key: str) -> None:
+    """Render currently playing users/items and transcoding state."""
+    st.subheader("Now playing")
+    try:
+        sessions = cached_active_sessions(base_url, api_key)
+    except Exception as exc:
+        st.warning(f"Could not load active sessions: {exc}")
+        return
+
+    if not sessions:
+        st.caption("No active playback sessions right now.")
+        return
+
+    rows = []
+    for session in sessions:
+        item = session.get("NowPlayingItem") or {}
+        session_id = session.get("Id") or ""
+        user_name = session.get("UserName") or "Unknown"
+        device = session.get("DeviceName") or session.get("Client") or ""
+        play_state = session.get("PlayState") or {}
+        paused = bool(play_state.get("IsPaused"))
+        state_label = "paused" if paused else "playing"
+
+        series = item.get("SeriesName") or ""
+        if series:
+            title = f"{series} - {item.get('Name', '')}"
+        else:
+            title = item.get("Name") or "Unknown"
+
+        transcoding = session.get("TranscodingInfo") or {}
+        is_transcoding = bool(transcoding)
+        transcode_type = []
+        if is_transcoding:
+            if transcoding.get("IsVideoDirect") is False:
+                transcode_type.append("video")
+            if transcoding.get("IsAudioDirect") is False:
+                transcode_type.append("audio")
+            if not transcode_type:
+                transcode_type.append("active")
+
+        rows.append(
+            {
+                "user": user_name,
+                "title": title,
+                "type": item.get("Type", ""),
+                "state": state_label,
+                "transcoding": "yes" if is_transcoding else "no",
+                "transcoding_type": ", ".join(transcode_type),
+                "device": device,
+                "session_id": session_id,
+            }
+        )
+
+    st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
+
+
+def render_resource_dashboard(get_ssh_client, ssh_args: tuple, media_root: str, detailed: bool = False) -> None:
+    """Render server resource summary or detailed resource charts."""
+    st.subheader("Server resource details" if detailed else "Server overview")
+    host, username, port, key_filename, password = ssh_args
+    ssh = get_ssh_client(host, username, port, key_filename, password)
+
+    try:
+        status = resource_collector_status(ssh)
+    except Exception as exc:
+        st.error(f"Could not check resource collector status: {exc}")
+        return
+
+    if detailed:
+        control_col, start_col, restart_col, stop_col, refresh_col = st.columns([2.3, 1, 1, 1, 1])
+        control_col.caption(f"Collector: `{status}` | sample interval: 10s | retention: 7 days / 70k samples")
+        if start_col.button("Start metrics", key="resource_start_metrics", use_container_width=True):
+            try:
+                st.success(start_resource_collector(ssh))
+            except Exception as exc:
+                st.error(str(exc))
+        if restart_col.button("Restart", key="resource_restart_metrics", use_container_width=True):
+            try:
+                st.success(restart_resource_collector(ssh))
+            except Exception as exc:
+                st.error(str(exc))
+        if stop_col.button("Stop metrics", key="resource_stop_metrics", use_container_width=True):
+            try:
+                st.info(stop_resource_collector(ssh))
+            except Exception as exc:
+                st.error(str(exc))
+        if refresh_col.button("Refresh", key="resource_refresh", use_container_width=True):
+            st.rerun()
+    else:
+        st.caption(f"Collector: `{status}`. Open the Resources tab for controls, diagnostics, and detailed charts.")
+
+    try:
+        rows = read_resource_metrics(ssh, max_lines=1000)
+    except Exception as exc:
+        st.error(f"Could not read resource metrics: {exc}")
+        rows = []
+
+    disk_path = media_root or "/"
+    try:
+        space = disk_space(ssh, disk_path)
+        used_pct_value = float(str(space.get("used_pct", "0")).rstrip("%") or 0)
+        disk_cols = st.columns(4)
+        disk_cols[0].metric("Disk used", human_size(space.get("used")))
+        disk_cols[1].metric("Disk available", human_size(space.get("available")))
+        disk_cols[2].metric("Disk total", human_size(space.get("size")))
+        disk_cols[3].metric("Used percent", f"{used_pct_value:.0f}%")
+        st.progress(min(max(used_pct_value / 100, 0), 1), text=f"{space.get('mount', disk_path)} on {space.get('filesystem', '')}")
+    except Exception as exc:
+        st.warning(f"Could not read disk space for {disk_path}: {exc}")
+
+    if not rows:
+        if detailed:
+            st.info("No resource history yet. Click 'Start metrics' and wait at least 10 seconds for the first sample. If this stays empty, use Restart to install the latest collector script.")
+            with st.expander("Collector diagnostics"):
+                try:
+                    st.code(resource_collector_debug_info(ssh))
+                except Exception as exc:
+                    st.error(f"Could not read collector diagnostics: {exc}")
+        else:
+            st.info("No resource history yet. Open the Resources tab to start the collector.")
+        return
+
+    df = pd.DataFrame(rows)
+    numeric_columns = [
+        "ts", "cpu_pct", "mem_pct", "net_rx_bytes_per_sec", "net_tx_bytes_per_sec", "disk_read_bps", "disk_write_bps"
+    ]
+    for column in numeric_columns:
+        if column in df.columns:
+            df[column] = pd.to_numeric(df[column], errors="coerce")
+    df = df.dropna(subset=["ts"])
+    df["time"] = pd.to_datetime(df["ts"], unit="s", utc=True).dt.tz_convert(None)
+
+    cutoff_ts = time.time() - 3600
+    all_sample_count = len(df)
+    df = df[df["ts"] >= cutoff_ts]
+    if df.empty:
+        st.info("No samples in the last hour yet.")
+        if detailed:
+            with st.expander("Resource sample diagnostics"):
+                if all_sample_count:
+                    raw_df = pd.DataFrame(rows)
+                    st.write(f"Parsed samples: {all_sample_count}")
+                    st.write(f"Newest remote sample age: {time.time() - float(raw_df['ts'].astype(float).max()):.0f} seconds")
+                    st.dataframe(raw_df.tail(10), use_container_width=True, hide_index=True)
+                else:
+                    st.write("No parseable samples found.")
+        return
+
+    df = df.sort_values("ts")
+    latest = df.iloc[-1]
+    avg_cpu = df["cpu_pct"].mean()
+    max_cpu = df["cpu_pct"].max()
+    avg_mem = df["mem_pct"].mean()
+    max_mem = df["mem_pct"].max()
+    avg_net_down = df["net_rx_bytes_per_sec"].mean()
+    max_net_down = df["net_rx_bytes_per_sec"].max()
+    avg_net_up = df["net_tx_bytes_per_sec"].mean()
+    max_net_up = df["net_tx_bytes_per_sec"].max()
+    avg_disk_read = df["disk_read_bps"].mean()
+    max_disk_read = df["disk_read_bps"].max()
+    avg_disk_write = df["disk_write_bps"].mean()
+    max_disk_write = df["disk_write_bps"].max()
+
+    metric_cols = st.columns(6)
+    metric_cols[0].metric("CPU now", f"{latest['cpu_pct']:.1f}%", f"avg {avg_cpu:.1f}% / peak {max_cpu:.1f}%")
+    metric_cols[1].metric("RAM now", f"{latest['mem_pct']:.1f}%", f"avg {avg_mem:.1f}% / peak {max_mem:.1f}%")
+    metric_cols[2].metric("Network down", format_rate_bytes(latest["net_rx_bytes_per_sec"]), f"avg {format_rate_bytes(avg_net_down)} / peak {format_rate_bytes(max_net_down)}")
+    metric_cols[3].metric("Network up", format_rate_bytes(latest["net_tx_bytes_per_sec"]), f"avg {format_rate_bytes(avg_net_up)} / peak {format_rate_bytes(max_net_up)}")
+    metric_cols[4].metric("Disk read", format_rate_bytes(latest["disk_read_bps"]), f"avg {format_rate_bytes(avg_disk_read)} / peak {format_rate_bytes(max_disk_read)}")
+    metric_cols[5].metric("Disk write", format_rate_bytes(latest["disk_write_bps"]), f"avg {format_rate_bytes(avg_disk_write)} / peak {format_rate_bytes(max_disk_write)}")
+
+    chart_df = df.set_index("time")
+    if detailed:
+        st.markdown("**CPU and RAM - last hour**")
+        st.line_chart(chart_df[["cpu_pct", "mem_pct"]], use_container_width=True)
+    else:
+        st.caption("Detailed CPU/RAM charts, network, disk I/O, raw samples, and collector controls are available in the Resources tab.")
+        return
+
+    net_down_df, net_down_suffix = scaled_rate_chart_df(chart_df, ["net_rx_bytes_per_sec"], ["download"])
+    net_up_df, net_up_suffix = scaled_rate_chart_df(chart_df, ["net_tx_bytes_per_sec"], ["upload"])
+    net_down_col, net_up_col = st.columns(2)
+    with net_down_col:
+        st.markdown(f"**Network down - last hour ({net_down_suffix})**")
+        st.line_chart(net_down_df, use_container_width=True)
+    with net_up_col:
+        st.markdown(f"**Network up - last hour ({net_up_suffix})**")
+        st.line_chart(net_up_df, use_container_width=True)
+
+    disk_read_df, disk_read_suffix = scaled_rate_chart_df(chart_df, ["disk_read_bps"], ["read"])
+    disk_write_df, disk_write_suffix = scaled_rate_chart_df(chart_df, ["disk_write_bps"], ["write"])
+    disk_read_col, disk_write_col = st.columns(2)
+    with disk_read_col:
+        st.markdown(f"**Disk read - last hour ({disk_read_suffix})**")
+        st.line_chart(disk_read_df, use_container_width=True)
+    with disk_write_col:
+        st.markdown(f"**Disk write - last hour ({disk_write_suffix})**")
+        st.line_chart(disk_write_df, use_container_width=True)
+
+    with st.expander("Raw resource samples"):
+        st.dataframe(df.sort_values("time", ascending=False), use_container_width=True, hide_index=True)
diff --git a/src/media_library_viewer/ui/file_browser.py b/src/media_library_viewer/ui/file_browser.py
new file mode 100644
index 0000000..bb892b7
--- /dev/null
+++ b/src/media_library_viewer/ui/file_browser.py
@@ -0,0 +1,271 @@
+"""SSH file browser UI."""
+
+from __future__ import annotations
+
+import json
+from pathlib import PurePosixPath
+from typing import Callable
+
+import pandas as pd
+import streamlit as st
+from st_aggrid import AgGrid, DataReturnMode, GridOptionsBuilder, GridUpdateMode, JsCode
+
+from media_library_viewer.utils import human_size, timestamp_to_local
+
+
+FILE_BROWSER_FILTER_KEYS = [
+    "file_browser_kind_filter",
+    "file_browser_search",
+    "file_browser_extension_filter",
+    "file_browser_sort",
+    "file_browser_descending",
+    "file_browser_page",
+]
+
+
+def reset_file_browser_filters() -> None:
+    """Clear directory-local filters before opening another folder."""
+    for key in FILE_BROWSER_FILTER_KEYS:
+        st.session_state.pop(key, None)
+
+
+def set_file_browser_path(path: str, selected_path: str | None = None, reset_filters: bool = True) -> None:
+    """Set current directory and selected path for the File browser."""
+    if reset_filters:
+        st.session_state["file_browser_reset_filters_pending"] = True
+    st.session_state["file_browser_current_dir"] = path
+    st.session_state["file_browser_selected_path"] = selected_path or path
+    st.session_state["file_browser_sync_path_input"] = True
+
+
+def aggrid_selected_rows(response: dict) -> list[dict]:
+    """Return selected rows from a streamlit-aggrid response."""
+    selected_rows = response.get("selected_rows")
+    if selected_rows is None:
+        return []
+    if isinstance(selected_rows, pd.DataFrame):
+        return selected_rows.to_dict("records")
+    return list(selected_rows)
+
+
+def render_file_browser(cached_dir_listing: Callable[..., list[dict]], ssh_args: tuple, initial_path: str) -> str:
+    """Render the File browser and return the selected file/folder path."""
+    host, username, port, key_filename, password = ssh_args
+    st.subheader("Remote filesystem")
+
+    if "file_browser_current_dir" not in st.session_state:
+        st.session_state["file_browser_current_dir"] = initial_path or "/"
+    if "file_browser_selected_path" not in st.session_state:
+        st.session_state["file_browser_selected_path"] = st.session_state["file_browser_current_dir"]
+    if st.session_state.pop("file_browser_reset_filters_pending", False):
+        reset_file_browser_filters()
+    if "file_browser_path_input" not in st.session_state or st.session_state.pop("file_browser_sync_path_input", False):
+        st.session_state["file_browser_path_input"] = st.session_state["file_browser_current_dir"]
+
+    current_dir = st.session_state["file_browser_current_dir"] or "/"
+    selected = st.session_state.get("file_browser_selected_path", current_dir)
+
+    status_col, selected_col = st.columns([1, 2])
+    status_col.caption(f"Current folder: `{current_dir}`")
+    selected_col.caption(f"Selected path: `{selected}`")
+
+    path_col, up_col, go_col, select_col, refresh_col = st.columns([5, 1, 1, 1.5, 1.5])
+    path_input = path_col.text_input("Remote path", key="file_browser_path_input", label_visibility="collapsed")
+    requested_path = path_input or "/"
+    if up_col.button("Up", key="file_browser_up", use_container_width=True):
+        set_file_browser_path(str(PurePosixPath(current_dir).parent))
+        st.rerun()
+    if go_col.button("Go", key="file_browser_go", use_container_width=True):
+        set_file_browser_path(requested_path)
+        st.rerun()
+    if select_col.button("Select folder", key="file_browser_select_current_folder", use_container_width=True):
+        if requested_path != current_dir:
+            set_file_browser_path(requested_path, requested_path)
+        else:
+            st.session_state["file_browser_selected_path"] = current_dir
+        st.rerun()
+    if refresh_col.button("Refresh", key="file_browser_refresh", use_container_width=True):
+        cached_dir_listing.clear()
+        if requested_path != current_dir:
+            set_file_browser_path(requested_path)
+        st.rerun()
+
+    try:
+        rows = cached_dir_listing(host, username, port, key_filename, password, current_dir)
+    except Exception as exc:
+        st.error(f"Could not list directory `{current_dir}`: {exc}")
+        return st.session_state.get("file_browser_selected_path")
+
+    display_rows = []
+    for row in rows:
+        kind = "dir" if row["type"] == "d" else "file"
+        name = row["name"]
+        extension = PurePosixPath(name).suffix.lower() if kind == "file" else ""
+        full_path = str(PurePosixPath(current_dir) / name)
+        display_rows.append(
+            {
+                "kind": kind,
+                "label": "[DIR]" if kind == "dir" else "[FILE]",
+                "name": name,
+                "display_name": f"{'[DIR]' if kind == 'dir' else '[FILE]'} {name}",
+                "extension": extension,
+                "size_bytes": int(row["size"]),
+                "size": "-" if kind == "dir" else human_size(row["size"]),
+                "mtime": float(row["mtime"]),
+                "modified": timestamp_to_local(row["mtime"]),
+                "path": full_path,
+            }
+        )
+
+    total_count = len(display_rows)
+    dir_count = sum(1 for r in display_rows if r["kind"] == "dir")
+    file_count = total_count - dir_count
+    total_file_size = sum(r["size_bytes"] for r in display_rows if r["kind"] == "file")
+    st.caption(
+        f"Entries: {total_count} | Directories: {dir_count} | Files: {file_count} | File size: {human_size(total_file_size)}"
+    )
+
+    with st.container(border=True):
+        filter_col, search_col, ext_col, sort_col, order_col, page_size_col = st.columns([1.1, 2.3, 1.1, 1.2, 1, 1])
+        kind_filter = filter_col.selectbox("Show", ["All", "Directories", "Files"], key="file_browser_kind_filter", label_visibility="collapsed")
+        search_term = search_col.text_input("Search", placeholder="Search names", key="file_browser_search", label_visibility="collapsed")
+        known_exts = sorted({r["extension"] for r in display_rows if r["extension"]})
+        extension_filter = ext_col.selectbox("Ext", ["All"] + known_exts, key="file_browser_extension_filter", label_visibility="collapsed")
+        sort_by = sort_col.selectbox("Sort", ["Name", "Kind", "Size", "Modified"], key="file_browser_sort", label_visibility="collapsed")
+        descending = order_col.toggle("Desc", value=False, key="file_browser_descending")
+        page_size = page_size_col.selectbox("Rows", [10, 25, 50, 100, 200], index=1, key="file_browser_page_size", label_visibility="collapsed")
+
+    filtered_rows = display_rows
+    if kind_filter == "Directories":
+        filtered_rows = [r for r in filtered_rows if r["kind"] == "dir"]
+    elif kind_filter == "Files":
+        filtered_rows = [r for r in filtered_rows if r["kind"] == "file"]
+    if search_term:
+        needle = search_term.lower()
+        filtered_rows = [r for r in filtered_rows if needle in r["name"].lower()]
+    if extension_filter != "All":
+        filtered_rows = [r for r in filtered_rows if r["extension"] == extension_filter]
+
+    sort_key_map = {
+        "Name": lambda r: (r["kind"] != "dir", r["name"].lower()),
+        "Kind": lambda r: (r["kind"], r["name"].lower()),
+        "Size": lambda r: (r["kind"] != "dir", r["size_bytes"]),
+        "Modified": lambda r: r["mtime"],
+    }
+    filtered_rows.sort(key=sort_key_map[sort_by], reverse=descending)
+
+    filtered_count = len(filtered_rows)
+    page_count = max(1, (filtered_count + page_size - 1) // page_size)
+    page_col, summary_col = st.columns([1, 5])
+    if st.session_state.get("file_browser_page", 1) > page_count:
+        st.session_state["file_browser_page"] = page_count
+    page_number = page_col.number_input("Page", min_value=1, max_value=page_count, value=1, step=1, key="file_browser_page")
+    start = (int(page_number) - 1) * page_size
+    end = start + page_size
+    page_rows = filtered_rows[start:end]
+    summary_col.caption(f"Showing {start + 1 if filtered_count else 0}-{min(end, filtered_count)} of {filtered_count} matching entries")
+
+    parent_path = str(PurePosixPath(current_dir).parent)
+    visible_rows = []
+    if current_dir != "/":
+        visible_rows.append(
+            {
+                "kind": "up",
+                "label": "[UP]",
+                "name": "..",
+                "display_name": "[UP] ..",
+                "extension": "",
+                "size_bytes": 0,
+                "size": "-",
+                "mtime": 0.0,
+                "modified": "",
+                "path": parent_path,
+            }
+        )
+    visible_rows.extend(page_rows)
+
+    if not display_rows:
+        st.info("Directory is empty.")
+    elif not page_rows:
+        st.info("No entries match the current filters.")
+
+    if not visible_rows:
+        with st.expander("File browser diagnostics"):
+            st.write(f"Current folder: `{current_dir}`")
+            st.write(f"Path input: `{requested_path}`")
+            st.write(f"Selected path: `{selected}`")
+            st.write(f"Raw entries returned by remote listing: {len(rows)}")
+        return selected
+
+    table_rows = [
+        {
+            "type": row["kind"],
+            "name": row["name"],
+            "ext": row["extension"],
+            "size": row["size"],
+            "modified": row["modified"],
+            "path": row["path"],
+        }
+        for row in visible_rows
+    ]
+    table_df = pd.DataFrame(table_rows)
+
+    grid_builder = GridOptionsBuilder.from_dataframe(table_df)
+    grid_builder.configure_default_column(editable=False, resizable=True, sortable=False, filter=False)
+    grid_builder.configure_column("type", width=90)
+    grid_builder.configure_column("name", flex=2)
+    grid_builder.configure_column("ext", width=90)
+    grid_builder.configure_column("size", width=120)
+    grid_builder.configure_column("modified", width=180)
+    grid_builder.configure_column("path", hide=True)
+    grid_builder.configure_selection(selection_mode="single", use_checkbox=False)
+    grid_options = grid_builder.build()
+    grid_options["rowSelection"] = {
+        "mode": "singleRow",
+        "checkboxes": False,
+        "headerCheckbox": False,
+        "enableClickSelection": True,
+    }
+    grid_options["suppressRowClickSelection"] = False
+    grid_options["suppressCellFocus"] = True
+    grid_options["onCellClicked"] = JsCode(
+        """
+        function(event) {
+            if (event && event.node) {
+                event.node.setSelected(true, true);
+            }
+        }
+        """
+    )
+
+    grid_response = AgGrid(
+        table_df,
+        gridOptions=grid_options,
+        height=min(360, 33 * (len(table_rows) + 1)),
+        fit_columns_on_grid_load=True,
+        data_return_mode=DataReturnMode.AS_INPUT,
+        update_mode=GridUpdateMode.SELECTION_CHANGED,
+        key="file_browser_grid",
+        theme="streamlit",
+        allow_unsafe_jscode=True,
+    )
+
+    selected_rows = aggrid_selected_rows(grid_response)
+    if selected_rows:
+        picked_row = selected_rows[0]
+        picked_path = picked_row.get("path")
+        picked_type = picked_row.get("type")
+        action_token = f"{picked_type}:{picked_path}"
+
+        # Open directories immediately on row select, including the [UP] row.
+        if picked_type in {"dir", "up"} and picked_path and picked_path != current_dir:
+            if st.session_state.get("file_browser_last_row_action") != action_token:
+                st.session_state["file_browser_last_row_action"] = action_token
+                set_file_browser_path(picked_path)
+                st.rerun()
+        elif picked_path:
+            st.session_state["file_browser_selected_path"] = picked_path
+            st.session_state["file_browser_last_row_action"] = action_token
+
+    st.caption("Select a directory row to open it (including [UP] ..). Select a file row to target metadata/jobs.")
+    return st.session_state.get("file_browser_selected_path", current_dir)
diff --git a/src/media_library_viewer/ui/library.py b/src/media_library_viewer/ui/library.py
new file mode 100644
index 0000000..a0afcc6
--- /dev/null
+++ b/src/media_library_viewer/ui/library.py
@@ -0,0 +1,51 @@
+"""Jellyfin library browser UI pieces."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pandas as pd
+import streamlit as st
+
+from media_library_viewer.domain.media import media_streams
+from media_library_viewer.utils import ticks_to_minutes
+
+
+def show_item_card(client, item: dict[str, Any]) -> None:
+    """Render a poster card in the Jellyfin library grid."""
+    try:
+        st.image(client.image_url(item["Id"]), use_container_width=True)
+    except Exception:
+        st.caption("No image")
+    st.markdown(f"**{item.get('Name', 'Untitled')}**")
+    meta = [item.get("Type", "")]
+    if item.get("ProductionYear"):
+        meta.append(str(item["ProductionYear"]))
+    minutes = ticks_to_minutes(item.get("RunTimeTicks"))
+    if minutes:
+        meta.append(f"{minutes} min")
+    st.caption(" - ".join([m for m in meta if m]))
+    if st.button("Open", key=f"open-{item['Id']}"):
+        st.session_state["selected_item_id"] = item["Id"]
+
+
+def show_item_detail(client, item: dict[str, Any]) -> None:
+    """Render detailed Jellyfin item metadata for the selected poster card."""
+    st.header(item.get("Name", "Untitled"))
+    left, right = st.columns([1, 2])
+    with left:
+        st.image(client.image_url(item["Id"]), use_container_width=True)
+    with right:
+        st.write(item.get("Overview") or "No overview.")
+        st.write("**Path:**", item.get("Path") or "Not exposed by Jellyfin")
+        st.write("**Genres:**", ", ".join(item.get("Genres", [])) or "-")
+        st.write("**Rating:**", item.get("CommunityRating") or "-")
+        st.write("**Official rating:**", item.get("OfficialRating") or "-")
+
+    streams = media_streams(item)
+    if streams:
+        st.subheader("Jellyfin media streams")
+        st.dataframe(pd.DataFrame(streams), use_container_width=True)
+
+    with st.expander("Raw Jellyfin JSON"):
+        st.json(item)
diff --git a/src/media_library_viewer/ui/media.py b/src/media_library_viewer/ui/media.py
new file mode 100644
index 0000000..95a7e71
--- /dev/null
+++ b/src/media_library_viewer/ui/media.py
@@ -0,0 +1,214 @@
+"""Media index tab UI."""
+
+from __future__ import annotations
+
+from pathlib import PurePosixPath
+from typing import Any, Callable
+
+import pandas as pd
+import streamlit as st
+from st_aggrid import AgGrid, DataReturnMode, GridOptionsBuilder, GridUpdateMode, JsCode
+
+from media_library_viewer.services.media_index import MediaIndex, build_media_index
+
+
+def aggrid_selected_rows(response: dict[str, Any]) -> list[dict[str, Any]]:
+    """Return selected rows from a streamlit-aggrid response."""
+    selected_rows = response.get("selected_rows")
+    if selected_rows is None:
+        return []
+    if isinstance(selected_rows, pd.DataFrame):
+        return selected_rows.to_dict("records")
+    return list(selected_rows)
+
+
+def format_elapsed(seconds: float | int | None) -> str:
+    if seconds is None:
+        return ""
+    seconds = float(seconds)
+    if seconds < 60:
+        return f"{seconds:.1f}s"
+    minutes = int(seconds // 60)
+    remainder = seconds % 60
+    if minutes < 60:
+        return f"{minutes}m {remainder:.0f}s"
+    hours = minutes // 60
+    minutes = minutes % 60
+    return f"{hours}h {minutes}m"
+
+
+def render_media_tab(
+    client,
+    user_id: str,
+    libraries: list[dict[str, Any]],
+    set_file_browser_path: Callable[[str, str | None, bool], None],
+) -> None:
+    """Render the SQLite-backed media inventory tab."""
+    st.subheader("Media inventory")
+    st.caption("SQLite-backed index for full-library sorting/filtering. File size/bitrate/HDR are based on Jellyfin media source metadata, not a full ffprobe scan.")
+
+    index = MediaIndex()
+    status = index.status()
+
+    status_col, build_col, refresh_col = st.columns([3.5, 1.2, 1])
+    if status.exists:
+        status_parts = [f"Index: {status.item_count:,} items"]
+        if status.updated_at_label:
+            status_parts.append(f"updated {status.updated_at_label}")
+        if status.build_duration_seconds is not None:
+            status_parts.append(f"last build took {format_elapsed(status.build_duration_seconds)}")
+        status_col.caption(" | ".join(status_parts))
+    else:
+        status_col.warning("No local media index yet. Build it to enable full-library sorting and filtering.")
+
+    if build_col.button("Build index", key="media_index_build", use_container_width=True):
+        with st.spinner("Building media index from Jellyfin. This can take a while for large libraries..."):
+            count = build_media_index(client, user_id, libraries, index)
+        st.success(f"Indexed {count:,} media items.")
+        st.rerun()
+    if refresh_col.button("Refresh", key="media_index_refresh", use_container_width=True):
+        st.rerun()
+
+    if not index.status().exists:
+        st.info("The Media tab uses a local SQLite index so sorting by size, bitrate, HDR, codec, season, and episode works across the whole library rather than just the current Jellyfin page.")
+        return
+
+    library_options = {lib["Name"]: lib["Id"] for lib in libraries}
+
+    filter_col, type_col, search_col, page_size_col, page_col = st.columns([1.9, 1.8, 2.4, 1.1, 1])
+    selected_libraries = filter_col.multiselect(
+        "Libraries",
+        list(library_options.keys()),
+        default=list(library_options.keys()),
+        key="media_inventory_libraries",
+    )
+    media_types = type_col.multiselect(
+        "Types",
+        ["Movie", "Episode", "Video"],
+        default=["Movie", "Episode"],
+        key="media_inventory_types",
+    )
+    search = search_col.text_input("Search", key="media_inventory_search")
+    page_size = page_size_col.selectbox("Rows", [50, 100, 250, 500], index=1, key="media_inventory_page_size")
+    page = page_col.number_input("Page", min_value=1, value=1, step=1, key="media_inventory_page")
+
+    sort_options = {
+        "Title": "title",
+        "Series": "series",
+        "Season": "season",
+        "Episode": "episode",
+        "Type": "type",
+        "Year": "year",
+        "Runtime": "runtime",
+        "Size": "size",
+        "Bitrate": "bitrate",
+        "HDR": "hdr",
+        "Video codec": "video",
+        "Resolution": "resolution",
+        "Date added": "date_added",
+        "Library": "library",
+        "Path": "path",
+    }
+    sort_col, order_col, hdr_col = st.columns([1.4, 1.1, 1.2])
+    sort_label = sort_col.selectbox("Sort", list(sort_options.keys()), key="media_inventory_sort")
+    sort_order_label = order_col.selectbox("Order", ["Ascending", "Descending"], key="media_inventory_sort_order")
+    hdr_filter = hdr_col.selectbox("HDR filter", ["All", "HDR only", "SDR/unknown only"], key="media_inventory_hdr_filter")
+
+    if not selected_libraries:
+        st.info("Select at least one library to show indexed media.")
+        return
+
+    rows, total = index.query(
+        library_ids=[library_options[name] for name in selected_libraries],
+        media_types=media_types or ["Movie", "Episode", "Video"],
+        search=search,
+        hdr_filter=hdr_filter,
+        sort_key=sort_options[sort_label],
+        sort_order=sort_order_label,
+        limit=int(page_size),
+        offset=(int(page) - 1) * int(page_size),
+    )
+
+    st.caption(f"Showing {len(rows)} of {total:,} indexed matching items. Sort and filters apply to the full local index.")
+
+    columns = [
+        "title", "series", "season", "episode", "type", "year", "runtime_min",
+        "size", "bitrate", "hdr", "video", "resolution", "date_added", "library", "path", "id",
+    ]
+    if not rows:
+        st.info("No media found for the current filters.")
+        return
+
+    table_df = pd.DataFrame(rows)[columns].fillna("")
+    selected_media_path = st.session_state.get("media_inventory_selected_path")
+
+    grid_builder = GridOptionsBuilder.from_dataframe(table_df)
+    grid_builder.configure_default_column(editable=False, resizable=True, sortable=False, filter=False)
+    grid_builder.configure_column("title", header_name="Title", flex=2)
+    grid_builder.configure_column("series", header_name="Series", flex=1.5)
+    grid_builder.configure_column("season", header_name="Season", width=95)
+    grid_builder.configure_column("episode", header_name="Episode", width=105)
+    grid_builder.configure_column("type", header_name="Type", width=100)
+    grid_builder.configure_column("year", header_name="Year", width=90)
+    grid_builder.configure_column("runtime_min", header_name="Runtime (min)", width=125)
+    grid_builder.configure_column("size", header_name="Size", width=120)
+    grid_builder.configure_column("bitrate", header_name="Bitrate", width=125)
+    grid_builder.configure_column("hdr", header_name="HDR", width=80)
+    grid_builder.configure_column("video", header_name="Video codec", width=120)
+    grid_builder.configure_column("resolution", header_name="Resolution", width=120)
+    grid_builder.configure_column("date_added", header_name="Date added", width=120)
+    grid_builder.configure_column("library", header_name="Library", width=140)
+    grid_builder.configure_column("path", header_name="Path", flex=2)
+    grid_builder.configure_column("id", hide=True)
+    grid_builder.configure_selection(selection_mode="single", use_checkbox=False)
+    grid_options = grid_builder.build()
+    grid_options["rowSelection"] = {
+        "mode": "singleRow",
+        "checkboxes": False,
+        "headerCheckbox": False,
+        "enableClickSelection": True,
+    }
+    grid_options["suppressRowClickSelection"] = False
+    grid_options["suppressCellFocus"] = True
+    grid_options["onCellClicked"] = JsCode(
+        """
+        function(event) {
+            if (event && event.node) {
+                event.node.setSelected(true, true);
+            }
+        }
+        """
+    )
+    grid_response = AgGrid(
+        table_df,
+        gridOptions=grid_options,
+        height=min(650, 35 * (len(rows) + 1)),
+        fit_columns_on_grid_load=True,
+        data_return_mode=DataReturnMode.AS_INPUT,
+        update_mode=GridUpdateMode.SELECTION_CHANGED,
+        key="media_inventory_grid",
+        theme="streamlit",
+        allow_unsafe_jscode=True,
+    )
+    selected_rows = aggrid_selected_rows(grid_response)
+    if selected_rows:
+        selected_media_path = selected_rows[0].get("path")
+        if selected_media_path:
+            st.session_state["media_inventory_selected_path"] = selected_media_path
+            # Auto-sync File browser location from Media row selection.
+            # Guarded by last-synced path to avoid reapplying on every rerun.
+            last_synced = st.session_state.get("media_inventory_last_synced_path")
+            if selected_media_path != last_synced:
+                set_file_browser_path(str(PurePosixPath(selected_media_path).parent), selected_media_path)
+                st.session_state["media_inventory_last_synced_path"] = selected_media_path
+
+    if selected_media_path:
+        st.caption(f"Selected media path: `{selected_media_path}` (File browser folder synced automatically)")
+    else:
+        st.caption("Select a table row to automatically sync its containing folder to the File browser tab.")
+
+    with st.expander("Notes"):
+        st.write(
+            "The Media tab now queries a local SQLite index, so sorting/filtering applies across the indexed library. "
+            "Rebuild the index after Jellyfin scans or metadata changes. Full ffprobe enrichment for every item can be added later as a background index extension."
+        )
diff --git a/src/media_library_viewer/ui/preview.py b/src/media_library_viewer/ui/preview.py
new file mode 100644
index 0000000..c7d941e
--- /dev/null
+++ b/src/media_library_viewer/ui/preview.py
@@ -0,0 +1,129 @@
+"""Selected-file preview and remote path tools UI."""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+import pandas as pd
+import streamlit as st
+
+from media_library_viewer.jobs import JOB_TEMPLATES, run_job
+from media_library_viewer.utils import (
+    ffprobe_format_summary,
+    is_known_video_file,
+    summarize_audio_streams,
+    summarize_streams,
+    summarize_subtitle_streams,
+    summarize_video_streams,
+)
+
+
+def render_ffprobe_sections(ffprobe_data: dict[str, Any]) -> None:
+    """Render ffprobe output in separate container/video/audio/subtitle sections."""
+    format_summary = ffprobe_format_summary(ffprobe_data)
+    video_rows = summarize_video_streams(ffprobe_data)
+    audio_rows = summarize_audio_streams(ffprobe_data)
+    subtitle_rows = summarize_subtitle_streams(ffprobe_data)
+
+    st.markdown("**Container**")
+    st.dataframe(pd.DataFrame([format_summary]), use_container_width=True, hide_index=True)
+
+    st.markdown("**Video**")
+    if video_rows:
+        st.dataframe(pd.DataFrame(video_rows), use_container_width=True, hide_index=True)
+    else:
+        st.caption("No video streams found.")
+
+    st.markdown("**Audio**")
+    if audio_rows:
+        st.dataframe(pd.DataFrame(audio_rows), use_container_width=True, hide_index=True)
+    else:
+        st.caption("No audio streams found.")
+
+    st.markdown("**Subtitles**")
+    if subtitle_rows:
+        st.dataframe(pd.DataFrame(subtitle_rows), use_container_width=True, hide_index=True)
+    else:
+        st.caption("No subtitle streams found.")
+
+
+def render_selected_file_preview(
+    ssh_args: tuple,
+    selected_path: str | None,
+    cached_ffprobe_preview: Callable[..., dict[str, Any]],
+) -> None:
+    """Run and render a blocking ffprobe preview for selected known video files."""
+    with st.container(border=True):
+        st.markdown("**Selected file preview**")
+        if not selected_path:
+            st.caption("Select a file to preview media metadata.")
+            return
+
+        st.caption(f"Path: `{selected_path}`")
+        if not is_known_video_file(selected_path):
+            st.caption("Automatic ffprobe preview runs for known video file extensions only.")
+            return
+
+        refresh_col, status_col = st.columns([1.2, 5])
+        if refresh_col.button("Reload preview", key="preview_reload", use_container_width=True):
+            cached_ffprobe_preview.clear()
+            st.rerun()
+
+        host, username, port, key_filename, password = ssh_args
+        try:
+            with st.spinner("Running ffprobe preview..."):
+                ffprobe_data = cached_ffprobe_preview(host, username, port, key_filename, password, selected_path)
+        except Exception as exc:
+            status_col.error(f"ffprobe failed: {exc}")
+            return
+
+        status_col.success("ffprobe preview loaded.")
+        render_ffprobe_sections(ffprobe_data)
+        with st.expander("Raw ffprobe JSON"):
+            st.json(ffprobe_data)
+
+
+def render_ssh_tools(
+    ssh,
+    ssh_args: tuple,
+    selected_path: str | None,
+    cached_ffprobe_preview: Callable[..., dict[str, Any]],
+) -> None:
+    """Render selected-path diagnostics and safe job templates."""
+    render_selected_file_preview(ssh_args, selected_path, cached_ffprobe_preview)
+    if not selected_path:
+        return
+
+    st.subheader("Disk metadata and jobs")
+    tabs = st.tabs(["ffprobe", "stat", "jobs"])
+
+    with tabs[0]:
+        if st.button("Run ffprobe on selected path", key="tools_run_ffprobe"):
+            try:
+                data = ssh.ffprobe_json(selected_path)
+                render_ffprobe_sections(data)
+                with st.expander("All streams table"):
+                    st.dataframe(pd.DataFrame(summarize_streams(data)), use_container_width=True, hide_index=True)
+                with st.expander("Raw ffprobe JSON"):
+                    st.json(data)
+            except Exception as exc:
+                st.error(str(exc))
+
+    with tabs[1]:
+        if st.button("Run stat", key="tools_run_stat"):
+            result = ssh.stat_path(selected_path)
+            st.code(result.stdout or result.stderr)
+
+    with tabs[2]:
+        st.warning("Jobs run commands on the remote server. Phase 1 includes safe/read-only templates only.")
+        job_key = st.selectbox("Job", list(JOB_TEMPLATES.keys()), format_func=lambda k: JOB_TEMPLATES[k].name)
+        st.caption(JOB_TEMPLATES[job_key].description)
+        command_preview = JOB_TEMPLATES[job_key].render({"path": selected_path})
+        st.code(command_preview, language="bash")
+        if st.button("Run selected job", key="tools_run_selected_job"):
+            result = run_job(ssh, job_key, selected_path)
+            st.write(f"Exit status: `{result.exit_status}`")
+            if result.stdout:
+                st.code(result.stdout)
+            if result.stderr:
+                st.error(result.stderr)
diff --git a/src/media_library_viewer/utils.py b/src/media_library_viewer/utils.py
new file mode 100644
index 0000000..a7fd412
--- /dev/null
+++ b/src/media_library_viewer/utils.py
@@ -0,0 +1,227 @@
+"""Formatting and ffprobe summarization helpers.
+
+These helpers are intentionally UI-framework independent. Streamlit renders the
+returned dictionaries/dataframes, but another frontend can reuse the same
+summaries.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from pathlib import PurePosixPath
+from typing import Any
+
+
+VIDEO_FILE_EXTENSIONS = {
+    ".3g2",
+    ".3gp",
+    ".avi",
+    ".divx",
+    ".flv",
+    ".m2ts",
+    ".m4v",
+    ".mkv",
+    ".mov",
+    ".mp4",
+    ".mpeg",
+    ".mpg",
+    ".mts",
+    ".ogm",
+    ".ogv",
+    ".rmvb",
+    ".ts",
+    ".vob",
+    ".webm",
+    ".wmv",
+}
+
+
+def ticks_to_minutes(ticks: int | None) -> int | None:
+    """Convert Jellyfin/Emby 100-nanosecond ticks to rounded minutes."""
+    if not ticks:
+        return None
+    return round(ticks / 10_000_000 / 60)
+
+
+def human_size(num: int | float | None) -> str:
+    """Format a byte count as B/KB/MB/GB/etc."""
+    if num is None:
+        return ""
+    value = float(num)
+    for unit in ["B", "KB", "MB", "GB", "TB", "PB"]:
+        if value < 1024 or unit == "PB":
+            return f"{value:.1f} {unit}" if unit != "B" else f"{int(value)} B"
+        value /= 1024
+    return f"{value:.1f} PB"
+
+
+def timestamp_to_local(ts: float | None) -> str:
+    if ts is None:
+        return ""
+    return datetime.fromtimestamp(ts).strftime("%Y-%m-%d %H:%M:%S")
+
+
+def is_known_video_file(path: str | None) -> bool:
+    """Return True when a path extension is one we should ffprobe automatically."""
+    if not path:
+        return False
+    return PurePosixPath(path).suffix.lower() in VIDEO_FILE_EXTENSIONS
+
+
+def format_duration(seconds: str | int | float | None) -> str:
+    if seconds in (None, ""):
+        return ""
+    try:
+        total = float(seconds)
+    except (TypeError, ValueError):
+        return str(seconds)
+    hours = int(total // 3600)
+    minutes = int((total % 3600) // 60)
+    secs = int(total % 60)
+    return f"{hours:02d}:{minutes:02d}:{secs:02d}"
+
+
+def format_bitrate(bit_rate: str | int | float | None) -> str:
+    if bit_rate in (None, ""):
+        return ""
+    try:
+        value = float(bit_rate)
+    except (TypeError, ValueError):
+        return str(bit_rate)
+    if value >= 1_000_000:
+        return f"{value / 1_000_000:.2f} Mbps"
+    if value >= 1_000:
+        return f"{value / 1_000:.0f} kbps"
+    return f"{value:.0f} bps"
+
+
+def _tags(stream: dict[str, Any]) -> dict[str, Any]:
+    return stream.get("tags") or {}
+
+
+def _disposition(stream: dict[str, Any], key: str) -> str:
+    value = (stream.get("disposition") or {}).get(key)
+    return "yes" if value == 1 else ""
+
+
+def _side_data_types(stream: dict[str, Any]) -> str:
+    values = []
+    for item in stream.get("side_data_list") or []:
+        if item.get("side_data_type"):
+            values.append(item["side_data_type"])
+    return ", ".join(values)
+
+
+def ffprobe_format_summary(ffprobe: dict[str, Any]) -> dict[str, str]:
+    """Summarize ffprobe container/format-level metadata."""
+    fmt = ffprobe.get("format") or {}
+    return {
+        "filename": fmt.get("filename", ""),
+        "format": fmt.get("format_name", ""),
+        "format_long": fmt.get("format_long_name", ""),
+        "duration": format_duration(fmt.get("duration")),
+        "size": human_size(float(fmt["size"])) if fmt.get("size") else "",
+        "bit_rate": format_bitrate(fmt.get("bit_rate")),
+        "stream_count": str(fmt.get("nb_streams", "")),
+    }
+
+
+def summarize_video_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
+    """Return video-only stream rows with video/HDR-related fields."""
+    rows = []
+    for stream in ffprobe.get("streams", []):
+        if stream.get("codec_type") != "video":
+            continue
+        tags = _tags(stream)
+        rows.append(
+            {
+                "index": stream.get("index"),
+                "codec": stream.get("codec_name"),
+                "profile": stream.get("profile"),
+                "resolution": f"{stream.get('width', '')}x{stream.get('height', '')}",
+                "pix_fmt": stream.get("pix_fmt"),
+                "bit_rate": format_bitrate(stream.get("bit_rate")),
+                "avg_fps": stream.get("avg_frame_rate"),
+                "color_range": stream.get("color_range"),
+                "color_space": stream.get("color_space"),
+                "color_transfer": stream.get("color_transfer"),
+                "color_primaries": stream.get("color_primaries"),
+                "side_data": _side_data_types(stream),
+                "language": tags.get("language"),
+                "title": tags.get("title"),
+                "default": _disposition(stream, "default"),
+            }
+        )
+    return rows
+
+
+def summarize_audio_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
+    """Return audio-only stream rows with channel/language/default fields."""
+    rows = []
+    for stream in ffprobe.get("streams", []):
+        if stream.get("codec_type") != "audio":
+            continue
+        tags = _tags(stream)
+        rows.append(
+            {
+                "index": stream.get("index"),
+                "codec": stream.get("codec_name"),
+                "profile": stream.get("profile"),
+                "channels": stream.get("channels"),
+                "layout": stream.get("channel_layout"),
+                "sample_rate": stream.get("sample_rate"),
+                "bit_rate": format_bitrate(stream.get("bit_rate")),
+                "language": tags.get("language"),
+                "title": tags.get("title"),
+                "default": _disposition(stream, "default"),
+                "forced": _disposition(stream, "forced"),
+            }
+        )
+    return rows
+
+
+def summarize_subtitle_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
+    """Return subtitle-only stream rows with language/forced/default fields."""
+    rows = []
+    for stream in ffprobe.get("streams", []):
+        if stream.get("codec_type") != "subtitle":
+            continue
+        tags = _tags(stream)
+        rows.append(
+            {
+                "index": stream.get("index"),
+                "codec": stream.get("codec_name"),
+                "codec_long": stream.get("codec_long_name"),
+                "language": tags.get("language"),
+                "title": tags.get("title"),
+                "default": _disposition(stream, "default"),
+                "forced": _disposition(stream, "forced"),
+                "hearing_impaired": _disposition(stream, "hearing_impaired"),
+            }
+        )
+    return rows
+
+
+def summarize_streams(ffprobe: dict[str, Any]) -> list[dict[str, Any]]:
+    rows = []
+    for stream in ffprobe.get("streams", []):
+        rows.append(
+            {
+                "index": stream.get("index"),
+                "type": stream.get("codec_type"),
+                "codec": stream.get("codec_name"),
+                "profile": stream.get("profile"),
+                "width": stream.get("width"),
+                "height": stream.get("height"),
+                "pix_fmt": stream.get("pix_fmt"),
+                "color_transfer": stream.get("color_transfer"),
+                "color_primaries": stream.get("color_primaries"),
+                "color_space": stream.get("color_space"),
+                "bit_rate": format_bitrate(stream.get("bit_rate")),
+                "channels": stream.get("channels"),
+                "sample_rate": stream.get("sample_rate"),
+                "language": stream.get("tags", {}).get("language"),
+                "title": stream.get("tags", {}).get("title"),
+            }
+        )
+    return rows
diff --git a/tests/.gitkeep b/tests/.gitkeep
new file mode 100644
index 0000000..e69de29