# Unified Ingest Autopilot Plan

**Objective:** Fully automatic end-to-end ingest pipeline — user drops files into a shared folder → orchestrator agent automatically claims, classifies, indexes into knowledge plane, extracts tasks, updates manifest, all without manual CLI commands. Must actually work now (code functional).

## Acceptance Criteria (testable)
1. Given a file in `~/UnifiedIngest/inbox/`, running `orchestrator claim_and_process` (or equivalent auto-trigger) results in: file moved to `sorted/<category>/`, manifest entry written (`ingest-manifest.jsonl`), index updated (`index/index.jsonl`), tasks extracted to `tasks/MASTER_TASKS.md` where applicable.
2. Duplicate detection: same SHA-256 → `duplicates/` folder + manifest `status: "duplicate"`.
3. Sensitive paths (`.ssh/`, `credentials.json`, names containing `password`/`secret`/`.key`, `.pem`) are skipped, recorded as `status: "skipped"` in manifest, and NOT passed to classification.
4. After ingest, `cyberbrain app.py --action search --query <term>` returns results from the ingested documents (semantic search over knowledge plane).
5. All existing tests pass: `pytest -q`, `ruff check`, `mypy src/cyberbrain26` (Gate 1 green baseline).
6. No unhandled exceptions for typical file types (.md, .txt, .pdf, .py, .jpg, .json, archives).

## Current State (codebase facts)
- **Unified ingest pipeline** already implemented:
  - `src/cyberbrain26/product/shared_root.py` — `resolve_shared_root`, `detect_syncthing_folders`, `INGEST_SUBDIRS` (inbox/processing/sorted/duplicates/index/tasks/manifest), `claim_batch`, `is_sensitive_path`, manifest read/write (append-only JSONL).
  - `src/cyberbrain26/product/ingest_worker.py` — `compute_sha256`, `classify_file` (categories: ai-export/note/doc/code/media/archive/other), `process_batch` (dedup against manifest, move to sorted/duplicates, manifest updates).
  - `src/cyberbrain26/product/index_builder.py` — `build_index` (generates `index.jsonl` prioritized by category), `extract_tasks` (extracts TODO/FIXME/ACTION/udělat/rozhodnuto/POTŘEBA lines into `tasks/MASTER_TASKS.md`).
  - CLI `cyberbrain app.py --action ingest --path <dir>` exists (commit `5739f3eb`) but uses `MemoryService.register_file` (knowledge plane), not the unified ingest pipeline.
- **Knowledge plane** `MemoryService.register_file` works per-file, extracts classification, chunks, embeds via Ollama `qwen3-embedding:0.6b` (dim=1024), stores in SQLite+LanceDB, navigates topics/tags/entity graph, audits every write.
- **Orchestrator agent** exists (`src/cyberbrain26/agents/orchestrator.py`) with bash exception allowed (`task: allow` for orchestrator only). Currently no wired commands to call `claim_batch`/`process_batch`/`build_index`/`extract_tasks`.
- **Tests** already present: `tests/test_ingest_worker.py`, `tests/test_index_builder.py`, `tests/test_independent_memory.py`, plus many `test_knowledge_*.py`. Gate 1 (747 tests, ruff, mypy) is green.

## Decisions (implementation choices)
1. **Pipeline foundation:** Use existing unified ingest pipeline (`shared_root.py` + `ingest_worker.py` + `index_builder.py`) as the core classification/organization layer. After `process_batch` classifies/moves files, optionally register into knowledge plane via `MemoryService.register_file` for each classified file (or keep separate; plan opts for dual-path: unified ingest for organization + knowledge plane for search).
2. **Orchestrator trigger:** Implement a **polling loop** (simple `while True: sleep 5; claim_and_process`) in the orchestrator agent, or a one-shot `orchestrator run_batch` CLI command. Polling is simpler and works without inotify deps. Decision: **polling** with configurable interval (env `INGEST_POLL_INTERVAL_SEC`, default 5).
3. **Shared root path:** Respect `UNIFIED_INGEST_ROOT` env, fallback to `~/UnifiedIngest`. Decision: **env-var driven**, no hard-coded paths.
4. **Knowledge plane integration:** After `process_batch` finishes, for each file moved to `sorted/<category>/`, run `MemoryService.register_file` (if category is note/doc/code/media; skip ai-export/other unless user wants). Decision: **conditional register** based on category.
5. **Sensitive handling:** Use existing `is_sensitive_path` from `shared_root.py`. Decision: **exact same logic**, skip + manifest entry.
6. **Error handling:** Per-file try/except in batch loop; errors logged to manifest `status: "error"` with reason; does NOT stop other files.

## Implementation Tasks (ordered for implementation-capable agent)
| # | Task | Description |
|---|------|-------------|
| 1 | **Add orchestrator CLI commands** `claim_and_process` + `run_ingest_cycle` in `src/cyberbrain26/agents/orchestrator.py` (or a new `ingest.py` module). Commands must: resolve shared root via `UNIFIED_INGEST_ROOT` env, run `claim_batch(limit=BATCH_LIMIT_MAX)`, then `process_batch(batch_id=...)`, then `build_index()`, then `extract_tasks()`. |
| 2 | **Implement polling loop** in orchestrator (or new `ingest_autopilot.py`) that every `INGEST_POLL_INTERVAL_SEC` seconds checks `inbox/` for new files and triggers `claim_and_process`. Must respect a `INGEST_ENABLE` env flag (default true). |
| 3 | **Wire knowledge plane register** after `process_batch`: for each file in `sorted/<category>/` where category ∈ {note, doc, code, media}, call `MemoryService.register_file(path, auto_classify=True)` (import from `cyberbrain26.knowledge_plane.memory_service`). Record document_id in manifest entry as `knowledge_doc_id`. |
| 4 | **Add robust error handling** in the orchestrator batch loop: catch `OSError`, `PermissionError`, embedding timeouts (Ollama retry), and manifest-write failures. Every error recorded in manifest with `status: "error"` + `reason`. |
| 5 | **Add unit/integration tests** in `tests/`:
   - `test_orchestrator_ingest_cycle.py` — end-to-end: create temp shared root, drop .md/.py/.jpg in inbox, run orchestrator cycle, assert: file → sorted/ note/doc/media, manifest entry `status: "classified"`, index entry, task extracted if applicable.
   - `test_ingest_duplicate_detection.py` — drop same file twice, assert second → duplicates/ + manifest `status: "duplicate"`.
   - `test_ingest_sensitive_skip.py` — drop file in `.ssh/` or named `passwords.json`, assert → skipped/ + manifest `status: "skipped"`.
   Run these with `pytest -q` and ensure they pass alongside existing suite.
| 6 | **Run Gate 1 verification**: `pytest -q`, `ruff check`, `mypy src/cyberbrain26`. Fix any new issues introduced.
| 7 | **Document usage** in `AGENTS.md` or `README.md`: how to set `UNIFIED_INGEST_ROOT`, `INGEST_POLL_INTERVAL_SEC`, `INGEST_ENABLE`; how to drop files; how to search results via `cyberbrain app.py --action search`.

## Risks & Mitigations
| Risk | Mitigation |
|------|------------|
| File race conditions (two agents claim same file) | `claim_batch` uses atomic `entry.replace` within same FS; orchestrator respects `limit` per batch; manifest tracks claimed SHA-256.
| Ollama embedding downtime | `_embed_with_retry` with exponential backoff + jitter (already in `batch_processor.py` pattern); orchestrator has per-file timeout + error → manifest `error`.
| False-positive sensitive-skip | `is_sensitive_path` uses substring matches on path string; tune `SENSITIVE_PATTERNS`/`SENSITIVE_EXTENSIONS` if needed; add allowlist if required.
| Index bloat / duplicate vector entries | `build_index` already deduplicates via SHA-256 hash; knowledge plane `MemoryService` has content-addressable embedding cache; manifest SHA-256 guards against re-register.
| Cross-device sync conflicts (Syncthing/rclone) | `detect_syncthing_folders` + `is_rclone_configured` already provide folder detection; orchestrator can read from any configured Syncthing folder path.

## Validation Plan
1. **Unit tests** (new): run `pytest -q tests/test_orchestrator_ingest_cycle.py tests/test_ingest_duplicate_detection.py tests/test_ingest_sensitive_skip.py`.
2. **Full suite**: `pytest -q` must still pass (747+ tests).
3. **Lint/type**: `ruff check` clean, `mypy src/cyberbrain26` clean (87 files).
4. **Manual E2E** (optional): set `UNIFIED_INGEST_ROOT=/tmp/test_ingest`, create `inbox/` with sample files (.md, .py, .jpg), run `orchestrator run_ingest_cycle`, verify outputs in `sorted/`, `index/`, `tasks/`, knowledge plane search `cyberbrain app.py --action search --query "TODO"`.
5. **Gate 2 readiness**: after above green, Gate 2 (post-R3 deployment comprehensive tests) can be triggered.

## Open Questions (to be decided by implementation agent or user)
- Should the orchestrator also register AI-export (`ai-export`) category into knowledge plane? (Currently plan says conditional on category; can be configured.)
- Desired default poll interval (default 5s; adjust based on system load).
- Should `extract_tasks` run on all categories or only note/doc? (Plan says note/doc primarily; can be configured via `TASK_PATTERNS` per category.)
- Whether to keep unified ingest pipeline **purely organizational** (folders, manifest) **and** knowledge plane **purely searchable**, or merge metadata (plan opts for dual-path with manifest linking).

---
*Plan generated 2026-11-04. See `.kilo/plans/1793832840086-unified-ingest-autopilot-plan.md` for the saved file.*