feat(system): skills validation, prompt architecture v2, pre-compact v3, context guardrail

Skills branch: 74→123 tests after 3-round validation, all commands operational, 99% seedgo. Prompt architecture: breadcrumbs pattern refined, devpulse git hygiene rewritten as natural habit. Pre-compact v3: injects STATUS.local.md + last session + git info before compaction. Global prompt: context guardrail added (flags off-topic conversations across terminals). Branch updates: daemon, backup, flow, memory, trigger, commons — code fixes from dispatch waves. Memory updates: .trinity/ and dev.local.md across all active branches. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-12 00:04:16 -07:00
parent fc0de2a76f
commit d3cabba738
66 changed files with 1614 additions and 882 deletions
@@ -43,7 +43,7 @@ src/aipass/{name}/

 ## Commands

-`drone` is a global CLI — never `cd` before running it. No full paths. Drone resolves everything.
+`drone` is a global CLI available in PATH. Never `cd` before running it. Never prefix with `export PATH=...` or full venv paths. Just `drone`. It resolves everything.

 ```
 drone @branch command [args]      # Route command to any branch
@@ -77,6 +77,10 @@ Prax is the ONLY logging system. Every branch uses:
 from aipass.prax import logger
 ```

+## Context Guardrail
+
+If the conversation suddenly shifts to a topic, project, or domain that doesn't relate to your current branch — **say something.** Don't just roll with it. Patrick uses voice input and multiple terminals. He may think he's talking to a different agent. A quick "Hey, this sounds like it's for [other project] — are you in the right terminal?" saves both of you from polluting memories with cross-context noise. Your job is to be the sanity check when the human has 5 windows open.
+
 ## Hard Rules

 - **No cross-branch file edits.** If you find an issue in another branch → email them.
@@ -98,6 +102,23 @@ Your `.trinity/` files are your persistence. Without them you're just an instanc

 Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`. If compaction hits before you save, it's gone. `dev.local.md` is for friction notes, ideas, and quick tracking — not formal docs. Details in your branch prompt.

+### Save Triggers — Do This Without Being Asked
+
+Save memories **proactively**. Don't wait for `/memo` or end of session. These are your triggers:
+- **After a milestone** — task completed, bug fixed, dispatch cycle done, plan closed
+- **After a decision** — Patrick chose an approach, rejected an idea, taught you something
+- **After learning something new** — a pattern, a gotcha, a command quirk, a system behavior
+- **Every ~10 turns** — if you haven't saved recently, save now. Context can compact at any time
+- **Before switching topics** — capture what you learned before the conversation moves on
+- **When Patrick teaches** — if he corrects you or shares insight, that's a key_learning immediately
+
+What to save where:
+- `local.json` → session entry (what happened), key_learnings (facts you'd need next time)
+- `observations.json` → collaboration patterns (how Patrick works, what works well, what to avoid)
+- `STATUS.local.md` → notepad for quick friction notes, completed items, issue tracking
+
+**The cost of saving too often is zero. The cost of losing context to compaction is everything.**
+
 ## Breadcrumbs

 Small knowledge traces that trigger awareness. Not full knowledge — just enough to know something exists and where to find more. A breadcrumb isn't the answer, it's the trigger that leads to the answer.
@@ -1,16 +1,118 @@
 #!/usr/bin/env python3
 """
-Pre-Compact Hook - Inject post-compact recovery instructions.
+Pre-Compact Hook - Inject live state for post-compact recovery.

-After compaction, Claude loses conversational flow and can come out
-disoriented. This hook injects recovery instructions into the summary
-to maintain conversational continuity.
+Reads STATUS.local.md, last session from local.json, and git branch
+to give the model real context after compaction — not generic advice.

-Version: 2.0.0
+Version: 3.0.0
 """

-import sys
 import json
+import subprocess
+import sys
+from pathlib import Path
+
+
+def _find_branch_dir():
+    """Find the current branch directory from CWD."""
+    cwd = Path.cwd()
+
+    # Check if we're in a branch dir or subdirectory of one
+    # Pattern: .../src/aipass/{branch}/...
+    parts = cwd.parts
+    for i, part in enumerate(parts):
+        if part == "aipass" and i > 0 and parts[i - 1] == "src":
+            branch_dir = Path(*parts[: i + 2])
+            if branch_dir.is_dir():
+                return branch_dir
+
+    # Check if CWD itself has .trinity/
+    if (cwd / ".trinity").is_dir():
+        return cwd
+
+    return None
+
+
+def _read_status_local(branch_dir):
+    """Read STATUS.local.md if it exists."""
+    for name in ["STATUS.local.md", "dev.local.md"]:
+        path = branch_dir / name
+        if path.is_file():
+            try:
+                return path.read_text(encoding="utf-8")[:3000]
+            except Exception:
+                pass
+    return None
+
+
+def _read_last_session(branch_dir):
+    """Read the most recent session and key_learnings from local.json."""
+    local_path = branch_dir / ".trinity" / "local.json"
+    if not local_path.is_file():
+        return None
+
+    try:
+        data = json.loads(local_path.read_text(encoding="utf-8"))
+        result = []
+
+        # Last session
+        sessions = data.get("sessions", [])
+        if sessions:
+            last = sessions[0]
+            result.append(
+                f"Last session (#{last.get('session_number', '?')}, "
+                f"{last.get('date', '?')}): {last.get('summary', 'no summary')}"
+            )
+
+        # Key learnings (just the keys, not full values — breadcrumbs)
+        learnings = data.get("key_learnings", {})
+        if learnings:
+            keys = list(learnings.keys())[-10:]  # last 10
+            result.append(f"Key learnings available: {', '.join(keys)}")
+
+        return "\n".join(result) if result else None
+    except Exception:
+        return None
+
+
+def _get_git_info():
+    """Get current git branch and short status."""
+    try:
+        branch = subprocess.run(
+            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        status = subprocess.run(
+            ["git", "diff", "--stat", "--cached", "HEAD"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+        dirty = subprocess.run(
+            ["git", "status", "--porcelain"],
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+
+        result = []
+        if branch.returncode == 0:
+            result.append(f"Git branch: {branch.stdout.strip()}")
+        if dirty.returncode == 0 and dirty.stdout.strip():
+            lines = dirty.stdout.strip().split("\n")
+            result.append(f"Uncommitted changes: {len(lines)} files")
+
+        return "\n".join(result) if result else None
+    except Exception:
+        return None
+
+
+def _get_branch_name(branch_dir):
+    """Extract branch name from directory."""
+    return branch_dir.name if branch_dir else "unknown"


 def main():
@@ -18,32 +120,46 @@ def main():
    try:
        json.load(sys.stdin)

-        print("Context compacting - recovery instructions injected", file=sys.stderr)
+        branch_dir = _find_branch_dir()
+        branch_name = _get_branch_name(branch_dir)

-        print("""POST-COMPACT RECOVERY (MANDATORY):
+        sections = []

-You just compacted. You're working from a summary now, not live memory.
+        sections.append(f"""POST-COMPACT RECOVERY — @{branch_name}

-STEP 1 — RE-ORIENT (do this silently, don't narrate):
- Read .trinity/local.json — your active tasks, recent sessions, key learnings
- Read .trinity/observations.json — collaboration patterns with Patrick
- Check .ai_mail.local/inbox.json — any unread mail?
- Your passport and system prompts are already injected via hooks — don't re-read those.
+Context just compacted. Below is your live state. Use it to continue seamlessly.""")

-STEP 2 — CONTINUE NATURALLY:
- Pick up where the conversation left off (the summary tells you where)
- Match the tone from before — if casual, stay casual
- Don't pivot to new topics or ask generic "what should we work on?" questions
- If mid-task: continue. If mid-discussion: continue. If uncertain: ask briefly.
+        # Git info
+        git_info = _get_git_info()
+        if git_info:
+            sections.append(f"## Git\n{git_info}")

-GUARDRAILS:
- Don't monologue about what you lost or what compaction did
- Don't surface old TODOs as fresh conversation starters
- Delegate code work to sub-agents — you're a manager, protect your context
- If about to debug/code in another branch's domain, delegate via ai_mail""", file=sys.stdout)
+        # Last session from local.json
+        if branch_dir:
+            session_info = _read_last_session(branch_dir)
+            if session_info:
+                sections.append(f"## Last Session\n{session_info}")

-    except Exception:
-        pass
+        # STATUS.local.md — the main context
+        if branch_dir:
+            status = _read_status_local(branch_dir)
+            if status:
+                sections.append(f"## STATUS.local.md\n{status}")
+
+        # Recovery instructions (lean)
+        sections.append("""## Recovery Protocol
+- Continue where the summary left off — don't restart or ask generic questions
+- STATUS.local.md Notepad has your scratch context
+- .trinity/local.json has full session history and key_learnings if you need more
+- Save memories proactively — compaction just proved you need to
+- Match the conversation tone from before compaction""")
+
+        print("\n\n".join(sections), file=sys.stdout)
+        print("Pre-compact: live state injected", file=sys.stderr)
+
+    except Exception as e:
+        # Fail silently — never block compaction
+        print(f"Pre-compact hook error: {e}", file=sys.stderr)

    sys.exit(0)

@@ -53,3 +53,7 @@ tools/
 test/
 src/aipass/seedgo/apps/standards/aipass/pack.json
 backup_data/
+
+
+backups
+backup_system
@@ -28,6 +28,8 @@ classifiers = [
 dependencies = [
    "rich>=13.0",
    "watchdog>=3.0",
+    "requests>=2.28",
+    "openai>=1.0",
 ]

 [project.urls]
@@ -1,14 +1,46 @@
-# API Branch-Local Context
-<!-- Source: /home/aipass/aipass_business/AIPass/src/aipass/api/.aipass/branch_system_prompt.md -->
+# API Branch — Local Context

-> Auto-created by aipass init. Customize for your branch.
+## Identity

-## Status: NEEDS CONFIGURATION
+API is the **LLM access layer** for AIPass. All branches route through here for model calls, key management, and usage tracking. Gateway to OpenRouter (and future providers).

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+## Commands

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+```
+drone @api get-key [provider]   # Retrieve API key (fallback: ~/.secrets/aipass/.env → config → env)
+drone @api validate [provider]  # Validate key format + connectivity
+drone @api test                 # Test OpenRouter connection
+drone @api models               # List available models
+drone @api track                # Track usage metrics
+drone @api stats                # Usage statistics
+```
+
+## Architecture
+
+3-tier: Entry point (`apps/api.py`) → Modules (3) → Handlers (9 files, 6 domains)
+
+**Modules:** `api_key.py` (key mgmt), `openrouter_client.py` (LLM client), `usage_tracker.py` (metrics)
+
+**Handlers:** `auth/` (keys, env), `config/` (provider), `openrouter/` (client, models, caller, provision), `usage/` (tracking, aggregation, cleanup), `json/` (auto-creating JSON ops)
+
+## Cross-Branch API
+
+```python
+from aipass.api.apps.modules.openrouter_client import get_response
+response = get_response(prompt="...", model="anthropic/claude-3.5-sonnet", caller="flow")
+```
+
+Used by: flow, prax, skills. Callers must specify model — no default (intentional).
+
+## Key Files
+
+- `apps/api.py` — Entry point with auto-discovery
+- `apps/handlers/auth/keys.py` — Key fallback chain (config → env → .env)
+- `apps/handlers/auth/env.py` — .env search paths (priority: `~/.secrets/aipass/`)
+- `api_json/` — Auto-created JSON storage (config, data, logs)
+
+## Memory & Tracking
+
+- `.trinity/` — Identity, session history, observations
+- `dev.local.md` — Working notes, todos, friction
+- `logs/` — Prax log output
@@ -38,6 +38,7 @@ from aipass.prax import logger
 # Navigate: env.py -> auth/ -> handlers/ -> apps/ -> api/
 API_ROOT = Path(__file__).resolve().parent.parent.parent.parent
 DEFAULT_ENV_PATHS = [
+    Path.home() / ".secrets" / "aipass" / ".env",  # ~/.secrets/aipass/.env (cross-platform standard)
    API_ROOT / ".env",                    # <api_root>/.env
    Path.cwd() / ".env",                  # <cwd>/.env
 ]
@@ -1,15 +1,22 @@
 # dev.local.md - API
 ```
-Branch: /home/coder/workspace/AIPass/src/aipass/api
+Branch: src/aipass/api
 Created: 2026-03-07
+Updated: 2026-03-10
 ```

 ## Issues

-
-
---
+- `track` command shows "TODO" — workflow not fully implemented in usage_tracker module
+- `make_call(args)`, `test_connection()`, `check_status()` in openrouter_client.py are TODO stubs
+- `list_providers()` in api_key.py hardcoded to "openrouter" only — should load dynamically

 ## Todos

-
+- Wire up real `track` workflow in usage_tracker.py
+- Implement `make_call()`, `test_connection()`, `check_status()` in openrouter_client.py
+- Consider logging import failures during module auto-discovery (currently silent)
+
+## Notes
+
+- 2026-03-10: Fixed missing `requests` + `openai` deps in pyproject.toml. All 6 commands now route correctly. Added `~/.secrets/aipass/.env` as priority search path. Configured branch prompt.
@@ -114,6 +114,16 @@ GLOBAL_IGNORE_PATTERNS = [
    ".gemini",  # AI assistant cache
    "snap",  # Snap package directories (broken symlinks, system-managed)

+    # AIPass internal state (runtime, not user data)
+    ".trinity",  # Branch identity/memory files (passport, local, observations)
+    ".ai_mail.local",  # Branch mailbox directories
+    ".archive",  # Branch archive directories
+    "backup_data",  # Backup runtime state
+    "backup_json",  # Backup metadata tracking
+    "DASHBOARD.local.json",  # Branch dashboard state
+    "CLOSED_PLANS.local.json",  # Flow plan close records
+    "STATUS.local.md",  # Branch status boards
+
    # Version control (huge number of files!)
    ".git",  # Git repositories are version controlled elsewhere

@@ -293,7 +303,6 @@ IGNORE_EXCEPTIONS = [
    # === MARKERS ===
    ".gitkeep",  # Include all .gitkeep marker files (especially in templates)
    ".gitattributes",  # Git attributes files
-    ".local.json"
 ]

 # Files that SHOULD have diffs created (exceptions to ignore patterns)
@@ -35,16 +35,10 @@ import datetime
 from pathlib import Path
 from typing import Dict

-from aipass.cli.apps.modules import console
+from aipass.cli.apps.modules import console, header, success, error
 from aipass.prax import logger


-def _header(text):
-    console.print(f"\n[bold cyan]{'='*70}[/bold cyan]")
-    console.print(f"[bold cyan]  {text}[/bold cyan]")
-    console.print(f"[bold cyan]{'='*70}[/bold cyan]")
-
-
 # Import handlers (core dependencies) - relative imports
 from aipass.backup.apps.handlers.config.config_handler import (
    BACKUP_MODES,
@@ -147,11 +141,6 @@ def handle_command(args) -> bool:

    Returns:
        bool: True if command was handled, False if not a backup command
-
-    Example:
-        result = handle_command(args)
-        if result:
-            console.print(f"Backup completed")
    """
    # Check if args has backup-related command
    if not hasattr(args, 'command'):
@@ -174,16 +163,41 @@ def handle_command(args) -> bool:
    # Get backup note if provided
    backup_note = getattr(args, 'note', 'No note provided')

+    # Rich header
+    header(f"Backup — {mode.title()}", {
+        'Mode': mode,
+        'Dry run': 'yes' if dry_run else 'no',
+    })
+
    try:
        # Create engine and run backup
        engine = BackupEngine(mode, dry_run=dry_run)
        result = engine.run_backup(backup_note)

-        # Return True to indicate command was handled
+        # Rich summary
+        duration = (datetime.datetime.now() - result.start_time).total_seconds()
+        console.print()
+        if result.critical_errors:
+            error(f"{mode.title()} backup FAILED")
+        elif result.errors > 0:
+            console.print(f"[yellow]{mode.title()} completed with {result.errors} errors[/yellow]")
+            console.print(f"  [dim]Files: {result.files_copied} copied, {result.files_skipped} skipped[/dim]")
+        else:
+            if dry_run:
+                success(f"{mode.title()} dry-run complete",
+                        files_scanned=result.files_checked,
+                        would_copy=result.files_copied,
+                        unchanged=result.files_skipped)
+            else:
+                success(f"{mode.title()} backup complete",
+                        files_copied=result.files_copied,
+                        files_skipped=result.files_skipped)
+        console.print(f"  [dim]Duration: {duration:.1f}s | Location: {result.backup_path}[/dim]")
+
        return True
    except Exception as e:
        logger.error(f"[backup_core] Backup command failed: {e}")
-        console.print(f"[ERROR] Backup failed: {e}")
+        error(f"Backup failed: {e}")
        return True  # Still handled, just failed


@@ -362,8 +376,7 @@ class BackupEngine:
            console.print("╰" + "─" * 68 + "╯", style="yellow")
            console.print()

-        _header(f"AIPass {self.mode_config['name']} - {self.mode_config['description']}")
-        console.print()
+        header(f"AIPass {self.mode_config['name']} - {self.mode_config['description']}")
        # Ensure backup directory exists
        if not self.ensure_backup_directory(result):
            console.print(f"\nBACKUP FAILED: Could not create backup directory")
@@ -2,14 +2,38 @@
 ```
 Branch: /home/patrick/Projects/AIPass/src/aipass/backup
 Created: 2026-03-07
+Last updated: 2026-03-10T14:15
 ```

 ## Issues

-
+- **reauth_drive.py uses old creds path** — `Path.home() / '.aipass' / 'drive_creds.json'` should be `Path.home() / '.secrets' / 'aipass' / 'drive_creds.json'`
+- **diff/ handlers not wired** — diff_generator.py, version_manager.py, vscode_integration.py exist but no CLI command routes to them. Dead code from UI perspective.
+- **integrations.py handle_command unreachable** — checks `args.integration_command` but CLI parser never sets this attribute. Route is dead.
+- **Seedgo 99%** — missing `logs/` and `dropbox/` directories per architecture template.
+- **Versioned backup exceeds drone timeout** — drone's 30s command timeout kills versioned backup when copying 16000+ files. Dry-run fits (10s), snapshot fits (14s). Need drone timeout increase or async handling.
+- **Dry-run updates timestamp** — `update_timestamp()` runs even in dry-run mode because `result.success` is True. Should skip timestamp update when `self.dry_run`.
+
+---
+
+## Fixed (2026-03-10, Round 2)
+
+- Added 9 AIPass-specific ignore patterns: `.trinity`, `.ai_mail.local`, `backup_data`, `backup_json`, `.archive`, `DASHBOARD.local.json`, `CLOSED_PLANS.local.json`, `STATUS.local.md`
+- Removed `.local.json` from IGNORE_EXCEPTIONS — was too broad, causing DASHBOARD/CLOSED_PLANS to leak through ignores
+- Verified: all patterns work, `dev.local.md` still backed up via `*.local.md` exception
+
+## Fixed (2026-03-10, Round 1)
+
+- Created `json_templates/default/` with config.json, data.json, log.json — was crashing `ensure_module_jsons`
+- Added `snap` to GLOBAL_IGNORE_PATTERNS in config_handler.py — broken symlinks in snap dirs crashed os.walk
+- Added `onerror` handler and `file_path.exists()` check in file_scanner.py — robustness for broken symlinks

 ---

 ## Todos

-
+- Wire diff/ handlers into CLI (add `diff` command to backup.py)
+- Fix integrations.py handle_command routing or remove dead path
+- Update reauth_drive.py creds path to ~/.secrets/aipass/
+- Create logs/ directory for 100% seedgo compliance
+- Set up Google Drive OAuth credentials when ready
@@ -1,15 +1,21 @@
 # dev.local.md - CLI
 ```
-Branch: /home/coder/workspace/AIPass/src/aipass/cli
+Branch: src/aipass/cli
 Created: 2026-03-07
 ```

 ## Issues

-
+- `handlers/display/` dir does NOT exist — display.py `print_introspection()` references it but shows "not found". Either create it or remove the reference.
+- `plugins/`, `extensions/`, `json_templates/` are empty stub packages from scaffold — consider if these are needed.
+- `__init__.py` top-level re-exports display functions but NOT templates (`operation_start`, `operation_complete`). README shows `from aipass.cli import operation_start` but that would fail.
+- Seedgo architecture at 91% — missing `.ai_mail.local/sent/`, `dropbox/`, `logs/` dirs.

 ---

 ## Todos

-
+- Fix top-level `__init__.py` to also re-export `operation_start`, `operation_complete` (or update README)
+- Create missing scaffold dirs for 100% seedgo architecture
+- Configure branch prompt (.aipass/aipass_local_prompt.md)
+- Consider creating handlers/display/ dir or removing reference from display.py introspection
@@ -1,14 +1,56 @@
-# DAEMON Branch-Local Context
-<!-- Source: /home/patrick/Projects/AIPass/src/aipass/daemon/.aipass/branch_system_prompt.md -->
+# DAEMON — Branch Context
+<!-- File: src/aipass/daemon/.aipass/aipass_local_prompt.md — Injected on every prompt when in daemon directory. -->

-> Auto-created by aipass init. Customize for your branch.
+Background scheduler and monitoring branch. Cron-triggered tasks, activity reports, action registry, scheduled follow-ups.

-## Status: NEEDS CONFIGURATION
+## Commands

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+```
+drone @daemon                          # Introspection — list discovered modules
+drone @daemon --help                   # Full help with all commands
+drone @daemon update                   # Status digest of daemon activity
+drone @daemon schedule list            # List pending scheduled tasks
+drone @daemon schedule create "task" --due 7d --to @branch
+drone @daemon schedule run-due         # Fire all due tasks (sends emails)
+drone @daemon activity                 # Quick 24h activity summary
+drone @daemon activity-report          # Full detailed report (--json for raw)
+drone @daemon branch-health BRANCH     # Single branch deep dive
+drone @daemon actions list             # Action registry
+drone @daemon actions set reminder 7d "msg" --to @branch
+drone @daemon actions set schedule @branch "prompt" daily 04:00
+```

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+Note: The `activity_report` module handles three commands: `activity`, `activity-report`, `branch-health`.
+
+## Apps Layout
+
+```
+apps/
+├── daemon.py              # Entry point — module discovery + command routing
+├── daemon_wakeup.py       # Wakeup / cron trigger
+├── scheduler_cron.py      # Cron scheduler
+├── modules/               # update, schedule, activity_report, actions, scheduler_ops, wakeup_ops
+├── handlers/
+│   ├── actions/           # actions_registry.py
+│   ├── json/              # json_handler.py
+│   ├── monitoring/        # activity_collector, memory_health, red_flag_detector, report_generator
+│   ├── schedule/          # task_registry, assistant_notifier, telegram_notifier
+│   ├── telegram/          # assistant_chat
+│   └── update/            # data_loader
+├── extensions/            # Extension point (empty)
+└── plugins/               # botfather_reminder, community_rotation, daily_audit, dev_central_monitor, heartbeat
+```
+
+## Known Issues
+
+- `activity_report` module shows as `activity_report` in `--help` but its actual commands are `activity`, `activity-report`, `branch-health` — calling `drone @daemon activity_report` fails
+- `branch-health` expects uppercase branch names from registry; lowercase fails
+- Secrets path: `~/.secrets/aipass/` (Path.home() / '.secrets' / 'aipass')
+
+## Memory & Tracking
+
+- `.trinity/passport.json` — identity
+- `.trinity/local.json` — session history
+- `.trinity/observations.json` — collaboration patterns
+- `dev.local.md` — scratchpad for issues, todos, notes
+- `DASHBOARD.local.json` — dashboard state
@@ -4,7 +4,7 @@
 **Module:** `aipass.daemon`
 **Created:** 2026-03-07
 **Citizen Class:** builder
-**Last Updated:** 2026-03-08
+**Last Updated:** 2026-03-10

 ---

@@ -36,7 +36,9 @@ daemon/
 │   │   ├── update.py          # Status digest module — summarizes DAEMON activity
 │   │   ├── schedule.py        # Scheduled follow-ups — fire-and-forget task management
 │   │   ├── activity_report.py # Branch activity report generator
-│   │   └── actions.py         # Action registry CLI — list, toggle, info, reminders
+│   │   ├── actions.py         # Action registry CLI — list, toggle, info, reminders
+│   │   ├── scheduler_ops.py   # Scheduler cron operations facade
+│   │   └── wakeup_ops.py      # Wake-up cron operations facade
 │   ├── handlers/
 │   │   ├── actions/
 │   │   │   └── actions_registry.py   # Action registry implementation
@@ -64,6 +66,8 @@ daemon/
 │       └── heartbeat.py               # Heartbeat / liveness plugin
 ├── daemon_json/                # JSON tracking data
 ├── docs/                       # Documentation
+├── dropbox/                    # Incoming file drops
+├── logs/                       # Prax log output
 ├── tools/                      # Branch verification utilities
 └── tests/                      # Test suite
 ```
@@ -98,6 +102,8 @@ drone @daemon <command> --help
 | `schedule` | Fire-and-forget scheduled follow-ups and task management |
 | `activity_report` | Branch activity report generator (plain text output) |
 | `actions` | Action registry CLI -- list, toggle, info, set reminder, set schedule, plugin migration |
+| `scheduler_ops` | Scheduler cron operations facade for scheduler_cron.py |
+| `wakeup_ops` | Wake-up cron operations facade for daemon_wakeup.py |

 ---

@@ -122,4 +128,4 @@ drone @daemon <command> --help

 ---

-*Last Updated: 2026-03-08*
+*Last Updated: 2026-03-10*
@@ -127,17 +127,18 @@ def print_help(modules: List[Any]):
    console.print("[bold cyan]AVAILABLE COMMANDS:[/bold cyan]")
    console.print()

-    if modules:
-        for module in modules:
-            module_name = module.__name__.split('.')[-1]
-            # Get first line of docstring
-            description = "No description"
-            if module.__doc__:
-                description = module.__doc__.strip().split('\n')[0]
+    # Show actual routable commands, not module names
+    _COMMAND_HELP = [
+        ("update", "Returns digest of DAEMON activity for check-ins."),
+        ("schedule", "CLI interface for fire-and-forget scheduled follow-ups."),
+        ("activity", "Quick 24-hour activity summary."),
+        ("activity-report", "Full detailed activity report (--json for raw)."),
+        ("branch-health", "Single branch deep dive (e.g., branch-health DAEMON)."),
+        ("actions", "CLI interface for the numbered action registry."),
+    ]

-            console.print(f"  [green]{module_name:20}[/green] [dim]{description}[/dim]")
-    else:
-        console.print("  [dim]No modules discovered[/dim]")
+    for cmd_name, desc in _COMMAND_HELP:
+        console.print(f"  [green]{cmd_name:20}[/green] [dim]{desc}[/dim]")

    console.print()
    console.print("-" * 70)
@@ -3,7 +3,7 @@
 # Description: DAEMON Wake-Up Cron Trigger
 # Version: 1.0.0
 # Created: 2026-02-15
-# Modified: 2026-02-15
+# Modified: 2026-03-10
 # =============================================

 """
@@ -13,10 +13,9 @@ Called periodically by cron. Standalone script -- not imported as a module.

 Flow:
  1. Acquire single-instance lock
-  2. Send Telegram "waking up" notification via daemon bot (optional)
-  3. Check daemon's email inbox (new/opened counts)
-  4. Build summary report with sender/subject listings
-  5. Send report notification via daemon bot (optional)
+  2. Check daemon's email inbox (new/opened counts)
+  3. Build summary report with sender/subject listings
+  4. Log report
 """

 # =============================================
@@ -26,33 +25,13 @@ Flow:
 import sys
 import json
 import fcntl
-import time
 from pathlib import Path
 from datetime import datetime

 from aipass.prax import logger
-# logger imported from aipass.prax

 from aipass.cli.apps.modules import console

-# =============================================
-# OPTIONAL IMPORTS (via module layer)
-# =============================================
-
-# Telegram notifications (optional) — route through modules, not handlers directly
-try:
-    from aipass.daemon.apps.modules.wakeup_ops import (
-        notify_wakeup,
-        notify_report,
-        notify_error,
-        TELEGRAM_AVAILABLE,
-    )
-except ImportError:
-    TELEGRAM_AVAILABLE = False
-    notify_wakeup = None
-    notify_report = None
-    notify_error = None
-
 # =============================================
 # CONSTANTS
 # =============================================
@@ -61,7 +40,6 @@ _DAEMON_ROOT = Path(__file__).resolve().parents[2]  # src/aipass/daemon/
 JSON_DIR = _DAEMON_ROOT / "daemon_json"

 LOCK_FILE = JSON_DIR / "wakeup.lock"
-CHAT_LOCK_FILE = JSON_DIR / "chat.lock"
 INBOX_PATH = _DAEMON_ROOT / "ai_mail.local" / "inbox.json"

 # =============================================
@@ -76,7 +54,7 @@ def print_introspection():
    console.print()
    console.print("Connected Handlers:")
    console.print("  modules/")
-    console.print("    - wakeup_ops.py (notify_wakeup, notify_report, notify_error — daemon bot Telegram notifications)")
+    console.print("    - wakeup_ops.py (notification stubs — Telegram stripped)")
    console.print()


@@ -98,32 +76,6 @@ def log(message: str) -> None:
    console.print(f"[{timestamp}] {message}")


-# =============================================
-# CHAT LOCK CHECK
-# =============================================
-
-def is_chat_active() -> bool:
-    """
-    Check if daemon is in an active Telegram chat session.
-
-    The chat listener creates a lock file while running.
-    If the lock exists and is fresh (<60 minutes), skip the wake-up cycle
-    to avoid interfering with the live conversation.
-
-    Returns:
-        True if chat session is active, False otherwise
-    """
-    if not CHAT_LOCK_FILE.exists():
-        return False
-    try:
-        age = time.time() - CHAT_LOCK_FILE.stat().st_mtime
-        if age > 3600:  # 60 minutes - treat as stale
-            return False
-        return True
-    except OSError:
-        return False
-
-
 # =============================================
 # EMAIL CHECK
 # =============================================
@@ -240,11 +192,6 @@ def main() -> int:
    log("=" * 60)
    log("Daemon wake-up triggered")

-    # Check if chat session is active (skip wake-up to avoid interference)
-    if is_chat_active():
-        log("Chat session active, skipping wake-up cycle")
-        return 0
-
    # Ensure lock directory exists
    LOCK_FILE.parent.mkdir(parents=True, exist_ok=True)

@@ -268,40 +215,18 @@ def _run_locked() -> int:
    """Execute the wake-up job while holding the lock."""
    exit_code = 0

-    # Step 1: Send "waking up" notification (optional)
-    if TELEGRAM_AVAILABLE:
-        try:
-            notify_wakeup()
-            log("Telegram: wakeup notification sent")
-        except Exception as e:
-            log(f"WARNING: Telegram wakeup notification failed: {e}")
-
-    # Step 2: Check inbox
+    # Step 1: Check inbox
    try:
        inbox = check_inbox()
        log(f"Inbox: {inbox['new_count']} new, {inbox['opened_count']} opened")
    except Exception as e:
        log(f"CRITICAL: Unhandled error in check_inbox: {e}")
-        if TELEGRAM_AVAILABLE:
-            try:
-                notify_error(f"Inbox check failed: {e}")
-            except Exception:
-                pass
        return 1

-    # Step 3: Build report
+    # Step 2: Build report
    report = build_report(inbox)
    log(f"Report: {report.splitlines()[0]}")

-    # Step 4: Send report notification (optional)
-    if TELEGRAM_AVAILABLE:
-        try:
-            notify_report(report)
-            log("Telegram: report notification sent")
-        except Exception as e:
-            log(f"WARNING: Telegram report notification failed: {e}")
-            exit_code = 1
-
    log("Daemon wake-up finished")
    log("=" * 60)
    return exit_code
@@ -314,9 +239,4 @@ if __name__ == "__main__":
        # Last-resort catch -- never crash silently
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        console.print(f"[{timestamp}] FATAL: Unhandled exception: {e}")
-        if TELEGRAM_AVAILABLE:
-            try:
-                notify_error(f"FATAL: {e}")
-            except Exception:
-                pass
        sys.exit(1)
@@ -21,8 +21,14 @@ from datetime import datetime, timedelta
 from typing import Dict, List, Any, Optional


-# Constants
-REGISTRY_PATH = Path(os.environ.get('AIPASS_REGISTRY', Path.home() / '.aipass' / 'AIPASS_REGISTRY.json'))
+# Constants — find registry: env var > repo root > ~/.aipass/
+_REPO_ROOT = Path(__file__).resolve().parents[6]  # src/aipass/daemon/apps/handlers/monitoring -> repo root
+_REGISTRY_CANDIDATES = [
+    Path(os.environ.get('AIPASS_REGISTRY', '')),
+    _REPO_ROOT / 'AIPASS_REGISTRY.json',
+    Path.home() / '.aipass' / 'AIPASS_REGISTRY.json',
+]
+REGISTRY_PATH = next((p for p in _REGISTRY_CANDIDATES if p.name and p.exists()), _REGISTRY_CANDIDATES[-1])
 MEMORY_FILE_PATTERNS = [".local.json", ".observations.json", "README.md"]
 CODE_FILE_EXTENSION = ".py"

@@ -1,164 +1,35 @@
 # =================== AIPass ====================
 # Name: assistant_notifier.py
-# Description: Daemon Bot Telegram Notifications
-# Version: 1.0.0
+# Description: Daemon Bot Notifications (Telegram stripped)
+# Version: 2.0.0
 # Created: 2026-02-15
-# Modified: 2026-02-15
+# Modified: 2026-03-10
 # =============================================

 """
-Handler for sending Telegram notifications via the daemon bot.
+Daemon bot notification stubs.

-Patrick's direct line to daemon. Uses the dedicated daemon bot
-(separate from scheduler and bridge bots) to notify Patrick of
-wake-up events, email reports, and errors.
+Telegram was stripped from daemon. These stubs remain so existing
+imports don't break. Will be replaced by a skill-based notification
+system later.
 """

-import os
-import json
-from datetime import datetime
-from pathlib import Path
-from typing import Dict
-from urllib.request import Request, urlopen
-from urllib.error import URLError
-
 from aipass.prax import logger
-# logger imported from aipass.prax

-# =============================================
-# CONSTANTS
-# =============================================
-
-CONFIG_PATH = Path(os.environ.get('AIPASS_DAEMON_CONFIG', Path.home() / '.aipass' / 'daemon_config.json'))
-
-
-# =============================================
-# CONFIG
-# =============================================
-
-def load_config() -> Dict[str, str]:
-    """
-    Load daemon bot config from daemon_config.json.
-
-    Returns:
-        Dict with 'bot_token' and 'chat_id' keys
-
-    Raises:
-        FileNotFoundError: If config file is missing
-        KeyError: If required keys are absent
-    """
-    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
-        raw = json.load(f)
-
-    return {
-        "bot_token": raw["telegram_bot_token"],
-        "chat_id": raw["telegram_chat_id"],
-    }
-
-
-# =============================================
-# SEND
-# =============================================
-
-def send_notification(message: str) -> bool:
-    """
-    Send a message to Patrick via the daemon bot.
-
-    Args:
-        message: Text to send (plain text, supports emoji)
-
-    Returns:
-        True if sent successfully, False otherwise
-    """
-    try:
-        config = load_config()
-    except (FileNotFoundError, KeyError, json.JSONDecodeError) as e:
-        logger.error(f"[daemon_notifier] Config error: {e}")
-        return False
-
-    url = f"https://api.telegram.org/bot{config['bot_token']}/sendMessage"
-    payload = {
-        "chat_id": config["chat_id"],
-        "text": message,
-    }
-
-    data = json.dumps(payload).encode("utf-8")
-    req = Request(url, data=data, headers={"Content-Type": "application/json"})
-
-    try:
-        with urlopen(req, timeout=15) as resp:
-            result = json.loads(resp.read())
-            if result.get("ok"):
-                return True
-            logger.error(f"[daemon_notifier] API error: {result.get('description')}")
-            return False
-    except URLError as e:
-        logger.error(f"[daemon_notifier] Send failed: {e}")
-        return False
-    except Exception as e:
-        logger.error(f"[daemon_notifier] Unexpected error: {e}")
-        return False
-
-
-# =============================================
-# NOTIFICATION HELPERS
-# =============================================

 def notify_wakeup() -> bool:
-    """
-    Notify that daemon is waking up.
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    now = datetime.now().strftime("%H:%M:%S")
-    message = "\U0001f916 Daemon waking up at " + now
-    return send_notification(message)
+    """Stub — Telegram removed."""
+    logger.info("[assistant_notifier] notify_wakeup() — no-op, Telegram stripped")
+    return False


 def notify_report(summary: str) -> bool:
-    """
-    Send daemon's wake-up report.
-
-    Args:
-        summary: Report content (email counts, listings, etc.)
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    message = "\U0001f4cb Daemon Report:\n" + summary
-    return send_notification(message)
+    """Stub — Telegram removed."""
+    logger.info("[assistant_notifier] notify_report() — no-op, Telegram stripped")
+    return False


 def notify_error(error: str) -> bool:
-    """
-    Notify that an error occurred during wake-up.
-
-    Args:
-        error: Error description
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    message = "\u274c Daemon Error:\n" + error
-    return send_notification(message)
-
-
-# =============================================
-# MAIN - Testing
-# =============================================
-
-if __name__ == "__main__":
-    print("assistant_notifier.py - manual test")
-    print(f"Config path: {CONFIG_PATH}")
-
-    try:
-        cfg = load_config()
-        print(f"Bot token: {cfg['bot_token'][:12]}...")
-        print(f"Chat ID: {cfg['chat_id']}")
-    except Exception as e:
-        print(f"Config load failed: {e}")
-
-    print("\nSending test notification...")
-    ok = send_notification("Test from assistant_notifier.py handler")
-    print(f"Result: {'OK' if ok else 'FAILED'}")
+    """Stub — Telegram removed."""
+    logger.info("[assistant_notifier] notify_error() — no-op, Telegram stripped")
+    return False
@@ -1,169 +1,35 @@
 # =================== AIPass ====================
 # Name: telegram_notifier.py
-# Description: DAEMON Scheduler Telegram Notifications
-# Version: 1.0.0
+# Description: DAEMON Scheduler Notifications (Telegram stripped)
+# Version: 2.0.0
 # Created: 2026-02-15
-# Modified: 2026-02-15
+# Modified: 2026-03-10
 # =============================================

 """
-Handler for sending Telegram notifications via the scheduler bot.
+Scheduler notification stubs.

-Reusable notification layer for scheduled events. Uses the dedicated
-scheduler bot (separate from bridge bot) to notify Patrick of
-triggered events, completions, and errors.
+Telegram was stripped from daemon. These stubs remain so existing
+imports don't break. Will be replaced by a skill-based notification
+system later.
 """

-import os
-import json
-from datetime import datetime
-from pathlib import Path
-from typing import Dict
-from urllib.request import Request, urlopen
-from urllib.error import URLError
-
 from aipass.prax import logger
-# logger imported from aipass.prax

-# =============================================
-# CONSTANTS
-# =============================================
-
-CONFIG_PATH = Path(os.environ.get('AIPASS_DAEMON_CONFIG', Path.home() / '.aipass' / 'daemon_config.json'))
-
-
-# =============================================
-# CONFIG
-# =============================================
-
-def load_config() -> Dict[str, str]:
-    """
-    Load scheduler bot config from daemon_config.json.
-
-    Returns:
-        Dict with 'bot_token' and 'chat_id' keys
-
-    Raises:
-        FileNotFoundError: If config file is missing
-        KeyError: If required keys are absent
-    """
-    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
-        raw = json.load(f)
-
-    return {
-        "bot_token": raw["telegram_bot_token"],
-        "chat_id": raw["telegram_chat_id"],
-    }
-
-
-# =============================================
-# SEND
-# =============================================
-
-def send_notification(message: str) -> bool:
-    """
-    Send a message to Patrick via the scheduler bot.
-
-    Args:
-        message: Text to send (plain text, supports emoji)
-
-    Returns:
-        True if sent successfully, False otherwise
-    """
-    try:
-        config = load_config()
-    except (FileNotFoundError, KeyError, json.JSONDecodeError) as e:
-        logger.error(f"[telegram_notifier] Config error: {e}")
-        return False
-
-    url = f"https://api.telegram.org/bot{config['bot_token']}/sendMessage"
-    payload = {
-        "chat_id": config["chat_id"],
-        "text": message,
-    }
-
-    data = json.dumps(payload).encode("utf-8")
-    req = Request(url, data=data, headers={"Content-Type": "application/json"})
-
-    try:
-        with urlopen(req, timeout=15) as resp:
-            result = json.loads(resp.read())
-            if result.get("ok"):
-                return True
-            logger.error(f"[telegram_notifier] API error: {result.get('description')}")
-            return False
-    except URLError as e:
-        logger.error(f"[telegram_notifier] Send failed: {e}")
-        return False
-    except Exception as e:
-        logger.error(f"[telegram_notifier] Unexpected error: {e}")
-        return False
-
-
-# =============================================
-# NOTIFICATION HELPERS
-# =============================================

 def notify_triggered(event_name: str) -> bool:
-    """
-    Notify that a scheduled event was triggered.
-
-    Args:
-        event_name: Name of the event that fired
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    now = datetime.now().strftime("%H:%M:%S")
-    message = f"\U0001f514 Scheduler: {event_name} triggered at {now}"
-    return send_notification(message)
+    """Stub — Telegram removed."""
+    logger.info(f"[telegram_notifier] notify_triggered({event_name}) — no-op, Telegram stripped")
+    return False


 def notify_complete(event_name: str, summary: str) -> bool:
-    """
-    Notify that a scheduled event completed successfully.
-
-    Args:
-        event_name: Name of the event that completed
-        summary: Brief summary of what happened
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    message = f"\u2705 Scheduler: {event_name} complete\n{summary}"
-    return send_notification(message)
+    """Stub — Telegram removed."""
+    logger.info(f"[telegram_notifier] notify_complete({event_name}) — no-op, Telegram stripped")
+    return False


 def notify_error(event_name: str, error: str) -> bool:
-    """
-    Notify that a scheduled event failed.
-
-    Args:
-        event_name: Name of the event that failed
-        error: Error description
-
-    Returns:
-        True if notification sent, False otherwise
-    """
-    message = f"\u274c Scheduler: {event_name} failed\n{error}"
-    return send_notification(message)
-
-
-# =============================================
-# MAIN - Testing
-# =============================================
-
-if __name__ == "__main__":
-    print("telegram_notifier.py - manual test")
-    print(f"Config path: {CONFIG_PATH}")
-
-    try:
-        cfg = load_config()
-        print(f"Bot token: {cfg['bot_token'][:12]}...")
-        print(f"Chat ID: {cfg['chat_id']}")
-    except Exception as e:
-        print(f"Config load failed: {e}")
-
-    print("\nSending test notification...")
-    ok = send_notification("Test from telegram_notifier.py handler")
-    print(f"Result: {'OK' if ok else 'FAILED'}")
+    """Stub — Telegram removed."""
+    logger.info(f"[telegram_notifier] notify_error({event_name}) — no-op, Telegram stripped")
+    return False
@@ -1,44 +1,21 @@
 # =================== AIPass ====================
 # Name: assistant_chat.py
-# Description: Daemon Bot Telegram Launcher
-# Version: 2.0.0
+# Description: Daemon Bot Telegram Launcher (STRIPPED)
+# Version: 3.0.0
 # Created: 2026-02-15
-# Modified: 2026-02-15
+# Modified: 2026-03-10
 # =============================================

 """
-Daemon Bot Telegram Launcher (@aipass_assistant_bot)
+Telegram chat launcher — STRIPPED.

-Thin wrapper around the shared direct_chat module. All chat logic,
-command handling, tmux management, and polling lives in direct_chat.py.
-This file only provides the daemon-specific configuration.
+Telegram is no longer part of daemon's responsibility.
+This file is kept as a placeholder. Will be replaced by
+a skill-based system later.
 """

 import sys
-from pathlib import Path

-from aipass.prax import logger
-# logger imported from aipass.prax
-
-try:
-    from api.apps.modules.telegram_chat import run_direct_chat
-    TELEGRAM_CHAT_AVAILABLE = True
-except ImportError:
-    TELEGRAM_CHAT_AVAILABLE = False
-    run_direct_chat = None
-
-_DAEMON_ROOT = Path(__file__).resolve().parents[3]  # src/aipass/daemon/
-
-if not TELEGRAM_CHAT_AVAILABLE:
-    logger.error("[assistant_chat] telegram_chat module not available, exiting")
-    sys.exit(1)
-
-sys.exit(run_direct_chat(
-    branch_name="daemon",
-    session_name="telegram-daemon",
-    config_path=Path.home() / ".aipass" / "assistant_bot_config.json",
-    work_dir=_DAEMON_ROOT,
-    log_dir=_DAEMON_ROOT / "logs",
-    data_dir=_DAEMON_ROOT / "daemon_json",
-    bot_name="AIPass Assistant Bot",
-))
+if __name__ == "__main__":
+    print("assistant_chat.py: Telegram stripped from daemon. No-op.")
+    sys.exit(0)
@@ -171,6 +171,13 @@ def handle_command(command: str, args: List[str]) -> bool:
    Returns:
        True if command was handled, False if not our command.
    """
+    # Handle 'activity_report' as alias — help shows module name, users expect it to work
+    if command == "activity_report":
+        hours = _parse_hours_arg(args)
+        report = generate_activity_report(since_hours=hours, verbosity="normal")
+        console.print(report)
+        return True
+
    # Handle 'activity' command - quick summary
    if command == "activity":
        if args and args[0] in ('--help', '-h', 'help'):
@@ -47,14 +47,26 @@ try:
    FILELOCK_AVAILABLE = True
 except ImportError:
    FILELOCK_AVAILABLE = False
+    FileLock = None  # type: ignore[assignment,misc]
+    Timeout = None  # type: ignore[assignment,misc]

-# Email integration (optional)
-try:
-    from ai_mail.apps.modules.email import send_email_direct
-    AI_MAIL_AVAILABLE = True
-except ImportError:
-    AI_MAIL_AVAILABLE = False
-    send_email_direct = None
+# Email integration via drone subprocess
+import subprocess
+
+def _send_email_via_drone(to_branch, subject, message, from_branch='@daemon',
+                          auto_execute=True, reply_to=None, **kwargs):
+    """Send email via drone @ai_mail send subprocess."""
+    cmd = ["drone", "@ai_mail", "send", to_branch, subject, message]
+    if auto_execute:
+        cmd.append("--dispatch")
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return result.returncode == 0
+    except (subprocess.SubprocessError, OSError):
+        return False
+
+AI_MAIL_AVAILABLE = True
+send_email_direct = _send_email_via_drone

 # =============================================
 # CONSTANTS
@@ -1,17 +1,17 @@
 # =================== AIPass ====================
 # Name: scheduler_ops.py
 # Description: Scheduler Cron Operations Module
-# Version: 1.0.0
+# Version: 2.0.0
 # Created: 2026-03-08
-# Modified: 2026-03-08
+# Modified: 2026-03-10
 # =============================================

 """
 Scheduler operations module -- facade for cron entry point.

 Provides a clean module-layer interface over handler functions
-used by scheduler_cron.py. Entry-level scripts import from
-this module instead of reaching into handlers directly.
+used by scheduler_cron.py. Telegram stripped — notification stubs
+remain for import compatibility.
 """

 from aipass.prax import logger
@@ -23,21 +23,14 @@ except ImportError:
    console = Console()

 # =============================================
-# TELEGRAM NOTIFICATIONS
+# NOTIFICATION STUBS (Telegram stripped)
 # =============================================

-try:
-    from aipass.daemon.apps.handlers.schedule.telegram_notifier import (
-        notify_triggered,
-        notify_complete,
-        notify_error,
-    )
-    TELEGRAM_AVAILABLE = True
-except ImportError:
-    TELEGRAM_AVAILABLE = False
-    notify_triggered = None
-    notify_complete = None
-    notify_error = None
+from aipass.daemon.apps.handlers.schedule.telegram_notifier import (
+    notify_triggered,
+    notify_complete,
+    notify_error,
+)

 # =============================================
 # TASK REGISTRY
@@ -96,7 +89,7 @@ def print_introspection():
    console.print()
    console.print("Connected Handlers:")
    console.print("  handlers/schedule/")
-    console.print("    - telegram_notifier.py (notify_triggered, notify_complete, notify_error — Telegram notifications)")
+    console.print("    - telegram_notifier.py (notification stubs — Telegram stripped)")
    console.print("    - task_registry.py (get_due_tasks, mark_dispatching, mark_completed, mark_pending, recover_stale_dispatches — task lifecycle)")
    console.print()
    console.print("  handlers/actions/")
@@ -109,24 +102,14 @@ def print_introspection():
 # =============================================

 def handle_command(command: str, args: list) -> bool:
-    """Handle commands routed by the entry point.
-
-    Args:
-        command: Command name
-        args: Additional arguments
-
-    Returns:
-        True if command was handled, False otherwise
-    """
+    """Handle commands routed by the entry point."""
    if command == "scheduler-ops":
-        from aipass.cli.apps.modules import console
-
        console.print()
        console.print("[bold cyan]Scheduler Ops[/bold cyan] - Cron operations facade")
        console.print()
-        console.print(f"  [dim]Telegram available:[/dim]  {TELEGRAM_AVAILABLE}")
-        console.print(f"  [dim]Task registry:[/dim]       {TASK_REGISTRY_AVAILABLE}")
-        console.print(f"  [dim]Action registry:[/dim]     {ACTION_REGISTRY_AVAILABLE}")
+        console.print(f"  [dim]Notifications:[/dim]    stubs (Telegram stripped)")
+        console.print(f"  [dim]Task registry:[/dim]    {TASK_REGISTRY_AVAILABLE}")
+        console.print(f"  [dim]Action registry:[/dim]  {ACTION_REGISTRY_AVAILABLE}")
        console.print()
        console.print("[dim]This module is a facade used by scheduler_cron.py.[/dim]")
        console.print()
@@ -1,17 +1,17 @@
 # =================== AIPass ====================
 # Name: wakeup_ops.py
 # Description: Wake-Up Cron Operations Module
-# Version: 1.0.0
+# Version: 2.0.0
 # Created: 2026-03-08
-# Modified: 2026-03-08
+# Modified: 2026-03-10
 # =============================================

 """
 Wake-up operations module -- facade for cron entry point.

 Provides a clean module-layer interface over handler functions
-used by daemon_wakeup.py. Entry-level scripts import from
-this module instead of reaching into handlers directly.
+used by daemon_wakeup.py. Telegram stripped — notification stubs
+remain for import compatibility.
 """

 from aipass.prax import logger
@@ -23,21 +23,14 @@ except ImportError:
    console = Console()

 # =============================================
-# DAEMON BOT TELEGRAM NOTIFICATIONS
+# NOTIFICATION STUBS (Telegram stripped)
 # =============================================

-try:
-    from aipass.daemon.apps.handlers.schedule.assistant_notifier import (
-        notify_wakeup,
-        notify_report,
-        notify_error,
-    )
-    TELEGRAM_AVAILABLE = True
-except ImportError:
-    TELEGRAM_AVAILABLE = False
-    notify_wakeup = None
-    notify_report = None
-    notify_error = None
+from aipass.daemon.apps.handlers.schedule.assistant_notifier import (
+    notify_wakeup,
+    notify_report,
+    notify_error,
+)


 # =============================================
@@ -48,11 +41,11 @@ def print_introspection():
    """Display module introspection info."""
    console.print()
    console.print("wakeup_ops Module")
-    console.print("Facade for daemon_wakeup.py — re-exports daemon bot Telegram notifications")
+    console.print("Facade for daemon_wakeup.py — notification stubs (Telegram stripped)")
    console.print()
    console.print("Connected Handlers:")
    console.print("  handlers/schedule/")
-    console.print("    - assistant_notifier.py (notify_wakeup, notify_report, notify_error — daemon bot Telegram notifications)")
+    console.print("    - assistant_notifier.py (notification stubs — Telegram stripped)")
    console.print()


@@ -61,22 +54,12 @@ def print_introspection():
 # =============================================

 def handle_command(command: str, args: list) -> bool:
-    """Handle commands routed by the entry point.
-
-    Args:
-        command: Command name
-        args: Additional arguments
-
-    Returns:
-        True if command was handled, False otherwise
-    """
+    """Handle commands routed by the entry point."""
    if command == "wakeup-ops":
-        from aipass.cli.apps.modules import console
-
        console.print()
        console.print("[bold cyan]Wakeup Ops[/bold cyan] - Cron wake-up facade")
        console.print()
-        console.print(f"  [dim]Telegram available:[/dim] {TELEGRAM_AVAILABLE}")
+        console.print("  [dim]Notifications:[/dim] stubs (Telegram stripped)")
        console.print()
        console.print("[dim]This module is a facade used by daemon_wakeup.py.[/dim]")
        console.print()
@@ -1,80 +1,31 @@
 # =================== AIPass ====================
 # Name: botfather_reminder.py
-# Description: Hourly reminder to create PATRICK_PRIVATE Telegram bot
-# Version: 1.0.0
+# Description: BotFather Reminder Plugin (DISABLED — Telegram stripped)
+# Version: 2.0.0
 # Created: 2026-02-26
-# Modified: 2026-02-26
+# Modified: 2026-03-10
 # =============================================

 """
-BotFather Reminder Plugin
+BotFather Reminder Plugin — DISABLED.

-Hourly reminder to create the PATRICK_PRIVATE Telegram bot via BotFather.
-Self-resolves: checks if bot token is still placeholder. Once real token
-is set in patrick_private.json, this plugin silently exits.
-
-Delete this file once the bot is created and running.
+Telegram was stripped from daemon. This plugin is no longer relevant.
+Kept as placeholder; will be removed once action registry entry is cleaned.
 """

-import json
-import subprocess
-from datetime import datetime
-from pathlib import Path
-
 PLUGIN_CONFIG = {
    "name": "botfather_reminder",
    "schedule": "hourly",
    "time": "00",
    "interval_minutes": None,
-    "enabled": True,
+    "enabled": False,
    "branch": "@dev_central",
    "fresh": False,
    "max_turns": 3,
-    "prompt": "Reminder: create PATRICK_PRIVATE Telegram bot via BotFather",
+    "prompt": "DISABLED — Telegram stripped from daemon",
 }

-BOT_CONFIG = Path.home() / ".aipass" / "telegram_bots" / "patrick_private.json"
-NOT_BEFORE = datetime(2026, 2, 27, 9, 0, 0)
-

 def run() -> dict:
-    """Check if reminder is still needed, send email if so."""
-    now = datetime.now()
-
-    # Don't start until BotFather cooldown expires
-    if now < NOT_BEFORE:
-        return {"status": "waiting", "reason": f"cooldown until {NOT_BEFORE.isoformat()}"}
-
-    # Check if token has been set (self-resolving)
-    try:
-        config = json.loads(BOT_CONFIG.read_text(encoding="utf-8"))
-        token = config.get("bot_token", "")
-        if token and token != "PASTE_TOKEN_HERE":
-            return {"status": "resolved", "reason": "token already set -- delete this plugin"}
-    except (json.JSONDecodeError, OSError):
-        pass  # Config missing or broken -- still remind
-
-    # Send reminder email
-    subject = "REMINDER: Create PATRICK_PRIVATE Telegram bot"
-    message = (
-        "BotFather cooldown should be expired. Steps:\n"
-        "1. Open Telegram -> @BotFather -> /newbot\n"
-        "2. Pick a name (anything private, e.g. 'My Notes')\n"
-        "3. Pick a username (must end in _bot)\n"
-        "4. Paste the token into the DEV_CENTRAL chat\n"
-        "5. Config ready at: ~/.aipass/telegram_bots/patrick_private.json\n\n"
-        "This reminder repeats hourly until the token is set."
-    )
-
-    try:
-        subprocess.run(
-            [
-                "drone", "@ai_mail", "send", "@dev_central",
-                subject, message,
-            ],
-            capture_output=True, timeout=15,
-        )
-    except (subprocess.SubprocessError, OSError):
-        pass  # Silent -- never crash scheduler
-
-    return {"status": "reminded", "next_check": "1 hour"}
+    """No-op — Telegram stripped."""
+    return {"status": "resolved", "reason": "Telegram stripped from daemon — plugin disabled"}
@@ -3,7 +3,7 @@
 # Description: DAEMON Scheduler Cron Trigger
 # Version: 2.0.0
 # Created: 2026-02-15
-# Modified: 2026-03-02
+# Modified: 2026-03-10
 # =============================================

 """
@@ -13,10 +13,10 @@ Called periodically by cron. Standalone script -- not imported as a module.

 Flow:
  1. Acquire single-instance lock
-  2. Send Telegram "triggered" notification (optional)
-  3. Recover stale dispatches
-  4. Process all due tasks (send emails, mark complete)
-  5. Send Telegram "complete" or "error" notification with summary (optional)
+  2. Recover stale dispatches
+  3. Process all due tasks (send emails, mark complete)
+  4. Process actions from registry
+  5. Log summary
 """

 # =============================================
@@ -41,20 +41,6 @@ from aipass.cli.apps.modules import console
 # OPTIONAL IMPORTS (via module layer)
 # =============================================

-# Resolve package imports — route through modules, not handlers directly
-try:
-    from aipass.daemon.apps.modules.scheduler_ops import (
-        notify_triggered,
-        notify_complete,
-        notify_error,
-        TELEGRAM_AVAILABLE,
-    )
-except ImportError:
-    TELEGRAM_AVAILABLE = False
-    notify_triggered = None
-    notify_complete = None
-    notify_error = None
-
 # Task registry (via module layer)
 try:
    from aipass.daemon.apps.modules.scheduler_ops import (
@@ -73,13 +59,21 @@ except ImportError:
    mark_pending = None
    recover_stale_dispatches = None

-# ai_mail (optional)
-try:
-    from ai_mail.apps.modules.email import send_email_direct
-    AI_MAIL_AVAILABLE = True
-except ImportError:
-    AI_MAIL_AVAILABLE = False
-    send_email_direct = None
+# Email integration via drone subprocess
+def _send_email_via_drone(to_branch, subject, message, from_branch='@daemon',
+                          auto_execute=True, reply_to=None, **kwargs):
+    """Send email via drone @ai_mail send subprocess."""
+    cmd = ["drone", "@ai_mail", "send", to_branch, subject, message]
+    if auto_execute:
+        cmd.append("--dispatch")
+    try:
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=15)
+        return result.returncode == 0
+    except (subprocess.SubprocessError, OSError):
+        return False
+
+AI_MAIL_AVAILABLE = True
+send_email_direct = _send_email_via_drone

 # Plugin discovery
 try:
@@ -137,9 +131,7 @@ def print_introspection():
    console.print()
    console.print("Connected Handlers:")
    console.print("  modules/")
-    console.print("    - scheduler_ops.py (notify_triggered, notify_complete, notify_error — Telegram notifications)")
-    console.print("    - scheduler_ops.py (get_due_tasks, mark_dispatching, mark_completed, mark_pending, recover_stale_dispatches — task registry ops)")
-    console.print("    - scheduler_ops.py (load_registry, is_action_due, update_last_run, mark_reminder_completed, migrate_plugins, next_due_str — action registry ops)")
+    console.print("    - scheduler_ops.py (task registry ops + action registry ops)")
    console.print()
    console.print("  plugins/")
    console.print("    - discover_plugins (plugin discovery and scheduled execution)")
@@ -822,27 +814,14 @@ def _run_locked() -> int:
    """Execute the cron job while holding the lock."""
    exit_code = 0

-    # Step 1: Send "triggered" notification (optional)
-    if TELEGRAM_AVAILABLE:
-        try:
-            notify_triggered(EVENT_NAME)
-            log("Telegram: triggered notification sent")
-        except Exception as e:
-            log(f"WARNING: Telegram triggered notification failed: {e}")
-
-    # Step 2: Process due tasks
+    # Step 1: Process due tasks
    try:
        results = process_due_tasks()
    except Exception as e:
        log(f"CRITICAL: Unhandled error in process_due_tasks: {e}")
-        if TELEGRAM_AVAILABLE:
-            try:
-                notify_error(EVENT_NAME, f"Unhandled error: {e}")
-            except Exception:
-                pass
        return 1

-    # Step 2.5: Process actions from registry (replaces old process_plugins)
+    # Step 2: Process actions from registry
    action_results = {
        "total": 0, "enabled": 0, "executed": 0, "failed": 0,
        "errors": [], "executed_actions": [], "skipped_actions": [],
@@ -853,7 +832,7 @@ def _run_locked() -> int:
        log(f"WARNING: Unhandled error in process_actions: {e}")
        action_results["errors"].append(f"Action processing: {e}")

-    # Step 3: Build detailed summary
+    # Step 3: Build summary
    lines = []

    # Tasks section
@@ -888,24 +867,9 @@ def _run_locked() -> int:

    log(f"Results: {summary}")

-    # Step 4: Send completion or error notification (optional)
-    if TELEGRAM_AVAILABLE:
-        try:
-            if results["failed"] > 0 or results["errors"] or action_results["failed"] > 0 or action_results["errors"]:
-                error_detail = summary
-                all_errors = results["errors"] + action_results["errors"]
-                if all_errors:
-                    error_detail += f"\nErrors:\n" + "\n".join(
-                        f"  - {e}" for e in all_errors[:5]
-                    )
-                notify_error(EVENT_NAME, error_detail)
-                log("Telegram: error notification sent")
-                exit_code = 1
-            else:
-                notify_complete(EVENT_NAME, summary)
-                log("Telegram: complete notification sent")
-        except Exception as e:
-            log(f"WARNING: Telegram result notification failed: {e}")
+    # Step 4: Determine exit code
+    if results["failed"] > 0 or results["errors"] or action_results["failed"] > 0 or action_results["errors"]:
+        exit_code = 1

    log("Scheduler cron finished")
    log("=" * 60)
@@ -919,9 +883,4 @@ if __name__ == "__main__":
        # Last-resort catch -- never crash silently
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        console.print(f"[{timestamp}] FATAL: Unhandled exception: {e}")
-        if TELEGRAM_AVAILABLE:
-            try:
-                notify_error(EVENT_NAME, f"FATAL: {e}")
-            except Exception:
-                pass
        sys.exit(1)
@@ -6,10 +6,25 @@ Created: 2026-03-07

 ## Issues

-
+- Memory health monitoring expects `DAEMON.local.json` (uppercase prefix) but actual files are `.trinity/local.json` — file naming mismatch in monitoring handlers
+- `branch-health` requires uppercase branch name from registry — `daemon` fails, needs `DAEMON`
+- Extensions dir is empty — no extensions implemented yet
+- Type errors in scheduler_cron.py (15) from optional imports with None fallback — pyright can't see runtime guards
+- Plugins vs actions registry redundancy — process_plugins() still exists alongside process_actions()
+
+## Resolved
+
+- ~~`activity_report` module routing mismatch~~ — FIXED: added `activity_report` as alias + help now shows actual commands
+- ~~`send_email_direct` phantom import~~ — FIXED: replaced with `_send_email_via_drone` subprocess wrapper
+- ~~AIPASS_REGISTRY.json path~~ — FIXED: activity_collector now checks repo root + ~/.aipass/
+- ~~Telegram code everywhere~~ — STRIPPED: 8 files cleaned, stubs remain for import compat

 ---

 ## Todos

-
+- Populate passport `what_i_do` / `what_i_dont_do` fields
+- Consolidate telegram notifier stubs into single file (or remove entirely once no callers remain)
+- Remove legacy process_plugins() once all plugins fully migrated to action registry
+- Fix memory_health to look for `.trinity/local.json` pattern instead of `{BRANCH}.local.json`
+- Secrets at `~/.secrets/aipass/` — note for Telegram bot token location (now stripped)
@@ -47,16 +47,32 @@ drone, seedgo, prax, cli, ai_mail, flow, spawn, trigger, api, backup, daemon, me

 - **Lean on branches.** You can't know everything — branches are the experts on their systems. When unsure, email them and ask. Don't burn context debugging what they already know.
 - **Use memories freely.** Don't hoard or stress about capacity — rollover to @memory is by design. Update `.trinity/` often. More is better.
- **dev.local.md for friction notes.** When something feels off or could be improved, drop a quick note. Address in batches later.
+- **STATUS.local.md for friction notes.** When something feels off or could be improved, drop a quick note in the Notepad section. Address in batches later.
 - **Know your limits.** You're great at planning, coordinating, seeing the big picture. You're bad at hands-on branch-level code tasks. Dispatch, don't do.
- **Git hygiene at breakpoints.** When a branch completes work, a plan closes, or a dispatch cycle finishes — run `git status` to see what's piling up. Don't let changes drift. Propose a commit or PR when it makes sense. Not every turn — just at natural milestones.
+- **Git awareness as a natural habit.** After completing a feature, merging something, or wrapping up a chunk of work — take a moment to think: "we've been working for a while, what's changed?" Run `git status`, see what's accumulated. If it looks like a coherent set of changes (an upgrade, a fix cycle, a config update), suggest a commit or PR. Don't force it every turn, but don't let 60+ files pile up silently either. Think of it like tidying your desk at the end of a work session — not obsessive, just mindful.
+
+## Autonomous Monitoring ⚡ LEARNING
+
+After dispatching branches, actively monitor for replies and keep them working without human intervention.
+
+**Option A: Task Agent** (costs tokens, gets reasoning)
+- Spawn a background agent that checks `drone @ai_mail inbox` periodically
+- Agent reads replies, evaluates quality, drafts responses
+- Use when results need judgment before re-dispatching
+
+**Option B: Bash Polling** (zero tokens, detection only)
+- Run `drone @ai_mail inbox 2>/dev/null | grep -q "📨"` in a background bash loop
+- Get notified when new mail arrives, then process manually
+- Use when just waiting for replies to come in
+
+**The loop:** Dispatch → monitor → read reply → respond/re-dispatch → monitor again. No human needed until you hit a decision point.

 ## Memory & Tracking

- `.trinity/local.json` — session history, active tasks, learnings
+- `.trinity/local.json` — session history, key learnings
 - `.trinity/observations.json` — collaboration patterns
- `dev.local.md` — issues, todos, working notes (human + AI shared scratchpad)
+- `STATUS.local.md` — current work, issues, todos, notepad (replaces dev.local.md). Feeds into central STATUS.md via `drone @prax status sync`.

-Update `.trinity/` proactively — after milestones, on `/memo`, at topic shifts, after 5+ actions without saving. Your persistence depends on it.
+Update `.trinity/` and `STATUS.local.md` proactively — after milestones, on `/memo`, at topic shifts, after 5+ actions without saving. Your persistence depends on it.

-**This prompt is NOT for tracking.** State goes in `.trinity/` and `dev.local.md`. This prompt = lightweight signposts injected every turn.
+**This prompt is NOT for tracking.** State goes in `.trinity/` and `STATUS.local.md`. This prompt = lightweight signposts injected every turn.
@@ -7,50 +7,47 @@ Updated: 2026-03-10

 ## Active Work

- **Flow dispatch**: DPLAN path fixes + FPLAN empty template detection — dispatched to flow, branch awake
- **Prompt architecture**: DONE — breadcrumbs in global+local, dev.local.md headers fixed system-wide
+- Nothing active — clean slate after FPLAN-0025 close

 ## Issues

- **flow/DPLAN**: All 10 handler files use Dev-Pass paths (`~/aipass_os/dev_central/dev_planning/`). Creates orphan dirs. Dispatched to flow.
- **flow/FPLAN**: `is_template_content()` too aggressive — deletes plans with real content if template boilerplate remains. Data loss bug. Dispatched to flow.
- **flow/FPLAN archive**: Closed FPLANs go to `src/aipass/backup_system/processed_plans/` — backup_system isn't a branch (Dev-Pass name). Leaving for later.
- **prax**: Missing `handlers/central/reader.py` — dashboard/refresh.py imports `read_all_centrals` but function never created
+- **flow/FPLAN archive**: FIXED by flow — now archives to flow/processed_plans/. Orphan backup_system/ removed.
+- **flow/DPLAN CWD default**: DPLANs always go to flow's dev_planning/ regardless of caller's CWD. Should default to current directory.
 - **backup**: Missing `config.json` template in json_templates/
 - **api**: `models` command not routed through drone
- **commons**: DB init failure (12/13 core branches operational)
+- **commons**: DB init failure (14/15 branches operational, commons only blocker)
 - **ai_mail**: `get_current_user()` returns relative `mailbox_path` — causes doubled paths in reply
- **drone**: stderr kwarg crash
+- **drone**: stderr kwarg crash on some error paths

 ## Completed

- PR #31 merged: seedgo v2, 589 files, full system audit (2026-03-10)
- ai_mail sender identity bug FIXED by Patrick
- Renamed flow.local.md → dev.local.md system-wide (17 files)
- Fixed dev.local.md headers system-wide (16 branch files + spawn template)
+- FPLAN-0025 STATUS board — built and closed (2026-03-10)
+  - STATUS.local.md seeded across 15 branches + spawn template
+  - Prax built sync handler autonomously (handlers/status/sync.py)
+  - drone @prax status sync verified: 14 operational, 1 in-progress
+  - Prax also fixed dashboard command collision + created central/reader.py
+- Claude Code statusline: ANSI colors, context bar, cost, LOC, hook activity flash
+- Hook logger pattern: ~/.claude/hook_logger.sh → /tmp/aipass-hook-last → statusline reads
+- Added Claude Code local docs breadcrumb to global prompt
+- PR #31 merged: seedgo v2, 589 files, full system audit
+- Prompt architecture: breadcrumbs in global+local, dev.local.md system-wide
+- Flow plan lifecycle: DPLAN paths fixed, FPLAN template detection fixed
 - FPLAN-0021 dispatched+completed by flow: DPLANs wired into CLI router
- Revised devpulse branch prompt — breadcrumb pattern, working habits, expertise table
- Added Breadcrumbs section + dev.local.md awareness to global prompt
- Updated .trinity/ memories with Session 18
- Processed inbox: ai_mail fix reply, flow FPLAN-0021 completion, old stress test
- Full FPLAN lifecycle tested: create (default+master), list, close (empty+content)
- Full DPLAN lifecycle tested: create, list, status, close

 ---

 ## Todos

- [x] Update global prompt with dev.local.md awareness
- [x] Fix dev.local.md headers system-wide
- [ ] Re-send emails to prax and backup about their known issues
 - [ ] Scaffold remaining modules with system prompts (.aipass/aipass_local_prompt.md)
- [ ] Test dispatch wake to branches beyond ai_mail
+- [ ] Test dispatch wake to branches beyond ai_mail/prax
 - [ ] Address drone stderr kwarg crash
 - [ ] Close stale FPLANs (0017, 0021) once verified complete
+- [ ] GWS CLI integration exploration for ai_mail external email transport

 ## Notepad

 - Dev-Pass reference: `/home/patrick/Projects/Dev-Pass/`
 - seed=Dev-Pass name, seedgo=AIPass name
- FPLAN close 5-step process: template check → mark closed → background archival → dashboard update → finalize
- DPLAN close 3-step process: close → background memory bank archival → finalize
+- STATUS board: `drone @prax status sync` rebuilds STATUS.md from all STATUS.local.md files
+- GWS CLI: `@googleworkspace/cli` — Rust-based, dynamic discovery. Future ai_mail integration candidate.
+- Claude Code API cost display in statusline — fun but not actionable (we don't use API directly)
@@ -1,5 +1,12 @@
 {
  "closed_plans": [
+    {
+      "plan_id": "DPLAN-002",
+      "type": "DPLAN",
+      "subject": "DPLAN-002__2026-03-10",
+      "date_closed": "2026-03-10",
+      "location": "dev_central"
+    },
    {
      "plan_id": "DPLAN-001",
      "type": "DPLAN",
@@ -9,12 +9,12 @@
 """
 Memory Bank Processing Handler

-Handles archival of closed PLAN files to backup_system/processed_plans/.
-AI summarization removed — plans vectorized directly from backup_system/processed_plans/.
+Handles archival of closed PLAN files to flow/processed_plans/.
+AI summarization removed — plans vectorized directly from flow/processed_plans/.

 Key Functions:
 - process_closed_plans() - Main entry point: archive plan → update registry
- archive_plan() - Move to backup_system/processed_plans/
+- archive_plan() - Move to flow/processed_plans/
 - is_template_content() - Template detection
 - verify_and_heal_orphaned_plans() - Orphan healing logic
 """
@@ -48,7 +48,7 @@ def _find_repo_root() -> Path:

 _REPO_ROOT = _find_repo_root()
 MEMORY_BANK_PATH = _REPO_ROOT / "MEMORY_BANK" / "plans"
-PROCESSED_PLANS_DIR = _PKG_ROOT / "backup_system" / "processed_plans"
+PROCESSED_PLANS_DIR = FLOW_ROOT / "processed_plans"
 PRIVATE_BRANCH_REGISTRY = _REPO_ROOT / "PRIVATE_BRANCH_REGISTRY.json"
 REGISTRY_FILE = FLOW_JSON_DIR / "flow_registry.json"
 CONFIG_FILE = FLOW_JSON_DIR / "flow_mbank_config.json"
@@ -347,7 +347,7 @@ def is_template_content(content: str) -> bool:

 # =============================================
 # CONTENT ANALYSIS (DISABLED)
-# AI summarization removed — plans vectorized directly from backup_system/processed_plans/
+# AI summarization removed — plans vectorized directly from flow/processed_plans/
 # =============================================

 # def analyze_plan_content(plan_path: Path) -> Dict[str, str]:
@@ -457,7 +457,7 @@ def is_template_content(content: str) -> bool:

 # =============================================
 # MEMORY BANK CREATION (DISABLED)
-# AI summarization removed — plans vectorized directly from backup_system/processed_plans/
+# AI summarization removed — plans vectorized directly from flow/processed_plans/
 # =============================================

 # def create_memory_entry(plan_path: Path, analysis: Dict[str, str]) -> Optional[Path]:
@@ -535,7 +535,7 @@ def is_template_content(content: str) -> bool:
 # =============================================

 def archive_plan(plan_path: Path) -> bool:
-    """Move processed plan file to backup_system/processed_plans/
+    """Move processed plan file to flow/processed_plans/

    VERIFICATION: Returns True ONLY if file successfully moved AND verified

@@ -739,7 +739,7 @@ def verify_and_heal_orphaned_plans() -> Dict[str, Any]:
 def process_closed_plans() -> Dict[str, Any]:
    """Main function to process all closed plans

-    # AI summarization removed — plans vectorized directly from backup_system/processed_plans/
+    # AI summarization removed — plans vectorized directly from flow/processed_plans/
    # Processing is now: archive_plan() → update registry flags → done

    AUTO-HEAL: Cleans up old -TEMP files from MEMORY_BANK after processing
@@ -780,7 +780,7 @@ def process_closed_plans() -> Dict[str, Any]:
                # Generate correlation ID for tracking
                correlation_id = f"FPLAN-{plan_num}-{datetime.now().strftime('%H%M%S')}"

-                # Archive plan to backup_system/processed_plans/
+                # Archive plan to flow/processed_plans/
                archive_success = archive_plan(plan_path)

                # Update registry flags
@@ -807,7 +807,7 @@ def process_closed_plans() -> Dict[str, Any]:
                    results.append({
                        "plan": f"FPLAN-{plan_num}",
                        "status": "archive_failed",
-                        "error": "Failed to move plan to backup_system/processed_plans/",
+                        "error": "Failed to move plan to flow/processed_plans/",
                        "correlation_id": correlation_id
                    })

@@ -53,7 +53,7 @@ def recover_plan_from_backup(plan_key: str, load_registry=None, save_registry=No
        (success, message)
    """
    # Check processed_plans directory
-    processed_plans = _PKG_ROOT / "backup_system" / "processed_plans"
+    processed_plans = FLOW_ROOT / "processed_plans"
    plan_file = processed_plans / f"FPLAN-{plan_key}.md"

    # CRITICAL: If base file doesn't exist, or if timestamp variants exist, use the NEWEST backup
@@ -35,7 +35,7 @@ MODULE_NAME = "post_close_runner"

 LOCK_FILE = FLOW_ROOT / ".post_close_runner.lock"

-# AI summarization removed — plans vectorized directly from backup_system/processed_plans/
+# AI summarization removed — plans vectorized directly from flow/processed_plans/
 # from aipass.flow.apps.handlers.summary.generate import generate_summaries
 from aipass.flow.apps.handlers.mbank.process import process_closed_plans

@@ -1,6 +1,6 @@
 {
  "DPLAN": {
-    "next_number": 2
+    "next_number": 3
  },
-  "next_number": 2
+  "next_number": 3
 }
@@ -1,14 +1,37 @@
 # MEMORY Branch-Local Context
-<!-- Source: /home/patrick/Projects/AIPass/src/aipass/memory/.aipass/branch_system_prompt.md -->

-> Auto-created by aipass init. Customize for your branch.
+## Identity

-## Status: NEEDS CONFIGURATION
+Memory is the central archive — vector search, rollover, and memory management for all AIPass branches. ChromaDB + sentence-transformers for semantic search. Rollover archives old `.trinity/` entries when files exceed 600 lines.

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+## Key Commands

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+```
+drone @memory search "query"        # Semantic search (requires torch)
+drone @memory search "q" --branch X # Filter by branch
+drone @memory rollover              # Execute rollover for triggered files
+drone @memory status                # Show rollover stats per branch
+drone @memory check                 # Dry run — what needs rollover
+drone @memory sync-lines            # Update line count metadata
+drone @memory watch                 # Auto-rollover watcher (Ctrl+C to stop)
+```
+
+## Architecture
+
+Entry point (`apps/memory.py`) auto-discovers modules via `handle_command()`. Two modules:
+- **rollover.py** — handles: rollover, status, check, sync-lines
+- **search.py** — handles: search
+
+Handlers implement domain logic under `apps/handlers/` (archive, json, learnings, monitor, rollover, schema, search, storage, tracking, vector).
+
+## Known Issues
+
+- `search` fails without `torch`/`sentence-transformers` installed
+- 5 commands in `--help` have no backing module: push-templates, diff-templates, template-status, symbolic demo, symbolic fragments
+- `status` shows 0 branches — may need registry path investigation
+
+## Memory & Tracking
+
+- `.trinity/` — passport, local.json, observations.json
+- `dev.local.md` — working scratchpad
+- `DASHBOARD.local.json` — system status
@@ -172,6 +172,40 @@ def _find_branches_near_rollover() -> List[Dict[str, Any]]:
                try:
                    data = json_loads(memory_file.read_text(encoding="utf-8"))
                    doc_meta = data.get("document_metadata", {})
+                    schema_version = doc_meta.get("schema_version", "1.0.0")
+                    limits = doc_meta.get("limits", {})
+
+                    # v2: check entry counts instead of line counts
+                    if schema_version.startswith("2"):
+                        max_sessions = limits.get("max_sessions")
+                        if max_sessions is not None:
+                            sessions = data.get("sessions", [])
+                            remaining_sessions = max_sessions - len(sessions)
+                            if remaining_sessions < 3:
+                                near_rollover.append({
+                                    "branch": branch_name,
+                                    "file_type": suffix,
+                                    "lines_remaining": remaining_sessions,
+                                    "current_lines": len(sessions),
+                                    "max_lines": max_sessions,
+                                    "v2_field": "sessions",
+                                })
+                        max_kl = limits.get("max_key_learnings")
+                        if max_kl is not None:
+                            kl = data.get("key_learnings", {})
+                            remaining_kl = max_kl - len(kl)
+                            if remaining_kl < 3:
+                                near_rollover.append({
+                                    "branch": branch_name,
+                                    "file_type": suffix,
+                                    "lines_remaining": remaining_kl,
+                                    "current_lines": len(kl),
+                                    "max_lines": max_kl,
+                                    "v2_field": "key_learnings",
+                                })
+                        continue
+
+                    # v1: line-count based
                    status = doc_meta.get("status", {})
                    current_lines = status.get("current_lines")

@@ -46,8 +46,12 @@ class RolloverTrigger:
    file_path: Path
    current_lines: int
    max_lines: int
+    schema_version: str = "1.0.0"
+    v2_reason: str = ""

    def __str__(self):
+        if self.schema_version.startswith("2") and self.v2_reason:
+            return f"{self.branch}.{self.memory_type} ({self.v2_reason})"
        return f"{self.branch}.{self.memory_type} ({self.current_lines}/{self.max_lines} lines)"


@@ -193,22 +197,62 @@ def _get_max_lines(file_path: Path, branch_name: str | None = None) -> int:
 # ROLLOVER DETECTION
 # =============================================================================

-def _should_rollover(file_path: Path) -> tuple[bool, int, int]:
+def _should_rollover(file_path: Path) -> tuple[bool, int, int, str, str]:
    """
-    Check if file should rollover
+    Check if file should rollover (supports v1 line-based and v2 entry-count based).

    Args:
        file_path: Path to memory JSON file

    Returns:
-        Tuple of (should_rollover, current_lines, max_lines)
+        Tuple of (should_rollover, current_lines, max_lines, schema_version, v2_reason)
+        For v2 files, max_lines is 0 and v2_reason describes which limits are exceeded.
    """
    current_lines = _count_file_lines(file_path)
-    max_lines = _get_max_lines(file_path)

-    should_trigger = current_lines >= max_lines
+    # Read file data once for schema detection + limit checks
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+    except Exception:
+        # Can't parse — fall back to line-based with hardcoded default
+        return (current_lines >= 600, current_lines, 600, '1.0.0', '')

-    return (should_trigger, current_lines, max_lines)
+    metadata = data.get('document_metadata', {})
+    schema_version = metadata.get('schema_version', '1.0.0')
+    limits = metadata.get('limits', {})
+
+    # v2: entry-count based limits
+    if schema_version.startswith('2'):
+        reasons = []
+
+        max_sessions = limits.get('max_sessions')
+        if max_sessions is not None:
+            sessions = data.get('sessions', [])
+            if isinstance(sessions, list) and len(sessions) > max_sessions:
+                reasons.append(f"{len(sessions)}/{max_sessions} sessions")
+
+        max_key_learnings = limits.get('max_key_learnings')
+        if max_key_learnings is not None:
+            key_learnings = data.get('key_learnings', {})
+            if isinstance(key_learnings, dict) and len(key_learnings) > max_key_learnings:
+                reasons.append(f"{len(key_learnings)}/{max_key_learnings} key_learnings")
+
+        max_observations = limits.get('max_observations')
+        if max_observations is not None:
+            observations = data.get('observations', [])
+            if isinstance(observations, list) and len(observations) > max_observations:
+                reasons.append(f"{len(observations)}/{max_observations} observations")
+
+        triggered = len(reasons) > 0
+        return (triggered, current_lines, 0, schema_version, ', '.join(reasons))
+
+    # v1: line-count based
+    max_lines = limits.get('max_lines')
+    if max_lines is None:
+        max_lines = _get_max_lines(file_path)
+
+    return (current_lines >= max_lines, current_lines, max_lines, '1.0.0', '')


 def check_all_branches() -> Dict[str, Any]:
@@ -244,7 +288,7 @@ def check_all_branches() -> Dict[str, Any]:
            if file_path is None:
                continue  # File doesn't exist, skip

-            should_trigger, current_lines, max_lines = _should_rollover(file_path)
+            should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)

            if should_trigger:
                trigger = RolloverTrigger(
@@ -252,7 +296,9 @@ def check_all_branches() -> Dict[str, Any]:
                    memory_type=memory_type,
                    file_path=file_path,
                    current_lines=current_lines,
-                    max_lines=max_lines
+                    max_lines=max_lines,
+                    schema_version=schema_ver,
+                    v2_reason=v2_reason,
                )
                triggers.append(trigger)

@@ -280,7 +326,7 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
            'error': f"File not found: {file_path}"
        }

-    should_trigger, current_lines, max_lines = _should_rollover(file_path)
+    should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)

    if should_trigger:
        # Extract branch and type from filename (e.g., SEED.observations.json)
@@ -293,7 +339,9 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
            memory_type=memory_type,
            file_path=file_path,
            current_lines=current_lines,
-            max_lines=max_lines
+            max_lines=max_lines,
+            schema_version=schema_ver,
+            v2_reason=v2_reason,
        )

        return {
@@ -302,12 +350,14 @@ def check_single_file(file_path: Path) -> Dict[str, Any]:
            'should_rollover': True
        }
    else:
+        remaining = max_lines - current_lines if max_lines > 0 else 0
        return {
            'success': True,
            'should_rollover': False,
            'current_lines': current_lines,
            'max_lines': max_lines,
-            'remaining': max_lines - current_lines
+            'schema_version': schema_ver,
+            'remaining': remaining
        }


@@ -344,14 +394,19 @@ def get_rollover_stats() -> Dict[str, Any]:
                continue

            stats['files_checked'] += 1
-            should_trigger, current_lines, max_lines = _should_rollover(file_path)
+            should_trigger, current_lines, max_lines, schema_ver, v2_reason = _should_rollover(file_path)

-            branch_stats[memory_type] = {
+            stat_entry = {
                'current': current_lines,
                'max': max_lines,
                'ready': should_trigger,
-                'remaining': max_lines - current_lines
+                'remaining': max_lines - current_lines if max_lines > 0 else 0,
+                'schema_version': schema_ver,
            }
+            if v2_reason:
+                stat_entry['v2_reason'] = v2_reason
+
+            branch_stats[memory_type] = stat_entry

            if should_trigger:
                stats['files_ready'] += 1
@@ -58,7 +58,11 @@ _startup_check_done = False

 def _get_rollover_threshold(branch_name: str, file_path: Path | None = None) -> int:
    """
-    Get rollover threshold for a memory file.
+    Get rollover threshold for a memory file (line-based, v1 only).
+
+    For v2 files (schema_version >= 2.0.0), returns a very large number so
+    line-based checks never trigger. v2 rollover is handled by the detector
+    using entry-count limits.

    Priority: file metadata > per_branch config > defaults > hardcoded 600

@@ -76,7 +80,14 @@ def _get_rollover_threshold(branch_name: str, file_path: Path | None = None) ->
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                data = json.load(f)
-                file_limit = data.get('document_metadata', {}).get('limits', {}).get('max_lines')
+                metadata = data.get('document_metadata', {})
+
+                # v2 files use entry-count limits, not line limits
+                schema_version = metadata.get('schema_version', '1.0.0')
+                if schema_version.startswith('2'):
+                    return 999999  # Never trigger line-based rollover for v2
+
+                file_limit = metadata.get('limits', {}).get('max_lines')
                if file_limit is not None:
                    return file_limit
        except Exception:
@@ -10,15 +10,18 @@
 Memory Extraction Handler

 Surgically extracts oldest items from memory files during rollover.
-Understands real JSON structure (sessions, observations arrays).
+Understands real JSON structure (sessions, observations arrays, key_learnings dict).

 Purpose:
-    When file exceeds 600 lines, extract oldest items from growing arrays
-    (sessions, observations, etc.), preserve JSON validity, update metadata.
+    v1 (schema <2.0.0): When file exceeds max_lines, extract oldest items from
+    growing arrays to get under line limit.
+    v2 (schema >=2.0.0): When entry counts exceed limits (max_sessions,
+    max_key_learnings), extract oldest entries by count.

 Strategy:
-    - Detect which array is growing (sessions, observations, etc.)
-    - Calculate how many items to remove to get under limit
+    - Detect schema version from document_metadata
+    - v1: line-count based extraction (legacy)
+    - v2: entry-count based extraction (sessions array + key_learnings dict)
    - Extract oldest items (FIFO)
    - Update document_metadata.status
 """
@@ -229,6 +232,100 @@ def _calculate_items_to_extract_by_lines(
    return max(1, total_items // 2)


+# =============================================================================
+# V2 EXTRACTION (ENTRY-COUNT BASED)
+# =============================================================================
+
+def _extract_items_v2(file_path: Path, data: Dict[str, Any]) -> Dict[str, Any]:
+    """
+    Extract items from v2 format file (entry-count based).
+
+    Handles sessions (array, oldest at end) and key_learnings (dict, oldest first).
+    Trims to max_sessions / max_key_learnings limits defined in document_metadata.
+
+    Args:
+        file_path: Path to memory JSON file
+        data: Already-parsed JSON data
+
+    Returns:
+        Dict with extracted items and metadata
+    """
+    limits = data.get('document_metadata', {}).get('limits', {})
+    old_lines = _count_file_lines(file_path)
+
+    all_extracted = []
+
+    # Extract from sessions array (newest first, oldest at end)
+    max_sessions = limits.get('max_sessions')
+    if max_sessions is not None:
+        sessions = data.get('sessions', [])
+        if isinstance(sessions, list) and len(sessions) > max_sessions:
+            excess = len(sessions) - max_sessions
+            extracted_sessions = sessions[-excess:]  # oldest from end
+            data['sessions'] = sessions[:-excess]    # keep newest
+            all_extracted.extend(extracted_sessions)
+
+    # Extract from key_learnings dict (first keys are oldest in insertion order)
+    max_key_learnings = limits.get('max_key_learnings')
+    if max_key_learnings is not None:
+        key_learnings = data.get('key_learnings', {})
+        if isinstance(key_learnings, dict) and len(key_learnings) > max_key_learnings:
+            excess = len(key_learnings) - max_key_learnings
+            keys_list = list(key_learnings.keys())
+            keys_to_extract = keys_list[:excess]  # oldest (first inserted)
+            for k in keys_to_extract:
+                all_extracted.append({'_type': 'key_learning', 'key': k, 'value': key_learnings[k]})
+                del data['key_learnings'][k]
+
+    # Extract from observations array (if v2 observations file)
+    max_observations = limits.get('max_observations')
+    if max_observations is not None:
+        observations = data.get('observations', [])
+        if isinstance(observations, list) and len(observations) > max_observations:
+            excess = len(observations) - max_observations
+            extracted_obs = observations[-excess:]
+            data['observations'] = observations[:-excess]
+            all_extracted.extend(extracted_obs)
+
+    if not all_extracted:
+        return {
+            'success': True,
+            'skipped': True,
+            'message': 'No entries exceed v2 limits'
+        }
+
+    # Update metadata
+    _update_metadata_after_extraction(data)
+
+    # Write back
+    try:
+        _write_memory_file(file_path, data)
+        new_lines = _count_file_lines(file_path)
+    except Exception as e:
+        return {
+            'success': False,
+            'error': f"Failed to write file: {e}"
+        }
+
+    # Parse branch and type from filename
+    parts = file_path.stem.split('.')
+    branch_name = parts[0] if len(parts) > 0 else "UNKNOWN"
+    memory_type = parts[1] if len(parts) > 1 else "unknown"
+
+    return {
+        'success': True,
+        'file': str(file_path),
+        'branch': branch_name,
+        'type': memory_type,
+        'array_field': 'v2_mixed',
+        'extracted': all_extracted,
+        'extracted_count': len(all_extracted),
+        'remaining_count': 0,
+        'old_lines': old_lines,
+        'new_lines': new_lines
+    }
+
+
 # =============================================================================
 # EXTRACTION OPERATIONS
 # =============================================================================
@@ -266,6 +363,12 @@ def extract_items(
            'error': f"Failed to read file: {e}"
        }

+    # v2 schema: delegate to entry-count based extraction
+    schema_version = data.get('document_metadata', {}).get('schema_version', '1.0.0')
+    if schema_version.startswith('2'):
+        return _extract_items_v2(file_path, data)
+
+    # v1: line-count based extraction
    # Detect structure
    array_field = _detect_growing_array(data)
    if not array_field:
@@ -176,8 +176,14 @@ def extract_text_from_memories(memories: List[Dict]) -> List[str]:
    for memory in memories:
        # Try common text fields
        if 'activities' in memory and isinstance(memory['activities'], list):
-            # Sessions type - join activities
+            # Sessions type (v1) - join activities
            text = '\n'.join(str(a) for a in memory['activities'])
+        elif 'summary' in memory:
+            # Sessions type (v2) - summary field
+            text = str(memory['summary'])
+        elif '_type' in memory and memory['_type'] == 'key_learning':
+            # Key learnings (v2) - key:value pair
+            text = f"{memory.get('key', '')}: {memory.get('value', '')}"
        elif 'content' in memory:
            text = str(memory['content'])
        elif 'text' in memory:
@@ -15,14 +15,10 @@ Fixes inconsistent schema in memory JSON files:
 3. Removes auto_compress_at (redundant with max_lines)
 4. Ensures document_metadata.status has current_lines

-Target schema:
-{
-  "document_metadata": {
-    "limits": { "max_lines": N, ... },
-    "status": { "current_lines": N, "last_health_check": "..." }
-  },
-  ... (other content)
-}
+Supports two schema versions:
+  v1 (schema_version <2.0.0): { "limits": { "max_lines": N } }
+  v2 (schema_version >=2.0.0): { "limits": { "max_sessions": N, "max_key_learnings": N,
+      "session_summary_max_chars": N, "learning_value_max_chars": N } }
 """

 import json
@@ -93,7 +89,10 @@ def normalize_memory_file(file_path: Path, dry_run: bool = False) -> Dict[str, A
        changes.append("Removed redundant 'auto_compress_at'")

    # 4. Remove unused limits fields (max_word_count, max_token_count - no code uses these)
+    # Preserve v2 fields: max_sessions, max_key_learnings, session_summary_max_chars, learning_value_max_chars
    if 'limits' in metadata:
+        v2_fields = {'max_sessions', 'max_key_learnings', 'session_summary_max_chars',
+                      'learning_value_max_chars', 'max_observations', 'max_lines', 'note'}
        for unused_field in ['max_word_count', 'max_token_count']:
            if unused_field in metadata['limits']:
                del metadata['limits'][unused_field]
@@ -98,15 +98,19 @@ def print_help() -> None:
    console.print("  python3 -m aipass.memory.apps.modules.rollover <command>")
    console.print()
    console.print("[bold]COMMANDS:[/bold]")
-    console.print("  [cyan]rollover[/cyan]    Execute rollover for files over 600 lines")
+    console.print("  [cyan]rollover[/cyan]    Execute rollover for files exceeding limits")
    console.print("  [cyan]status[/cyan]      Show rollover statistics for all branches")
    console.print("  [cyan]check[/cyan]       Check which files need rollover (dry run)")
    console.print("  [cyan]sync-lines[/cyan]  Update line count metadata for all branches")
    console.print("  [cyan]help[/cyan]        Show this help message")
    console.print()
+    console.print("[bold]LIMITS:[/bold]")
+    console.print("  v1 (schema <2.0): Line-count based (max_lines, default 600)")
+    console.print("  v2 (schema 2.0+): Entry-count based (max_sessions, max_key_learnings)")
+    console.print()
    console.print("[bold]WORKFLOW:[/bold]")
-    console.print("  1. Detect files over 600 lines")
-    console.print("  2. Extract oldest entries (target ~500 lines)")
+    console.print("  1. Detect files exceeding limits (line count or entry count)")
+    console.print("  2. Extract oldest entries")
    console.print("  3. Generate embeddings via sentence-transformers")
    console.print("  4. Store vectors in local + global ChromaDB")
    console.print()
@@ -253,16 +257,22 @@ def show_status() -> None:

            for memory_type, file_stats in branch_stats.items():
                current = file_stats['current']
-                max_lines = file_stats['max']
+                max_val = file_stats['max']
                ready = file_stats['ready']
                remaining = file_stats['remaining']
+                schema_ver = file_stats.get('schema_version', '1.0.0')
+                v2_reason = file_stats.get('v2_reason', '')

                status_marker = "[red]![/red]" if ready else "[green]OK[/green]"
-                status_text = "READY" if ready else f"{remaining} remaining"

-                console.print(
-                    f"    {status_marker} {memory_type}: {current}/{max_lines} lines ({status_text})"
-                )
+                if schema_ver.startswith('2') and v2_reason:
+                    status_text = f"READY ({v2_reason})" if ready else "OK (v2)"
+                    console.print(f"    {status_marker} {memory_type}: {status_text}")
+                else:
+                    status_text = "READY" if ready else f"{remaining} remaining"
+                    console.print(
+                        f"    {status_marker} {memory_type}: {current}/{max_val} lines ({status_text})"
+                    )

            console.print()

@@ -6,10 +6,15 @@ Created: 2026-03-07

 ## Issues

-
+- `search` command fails — missing `torch`/`sentence-transformers` deps
+- 5 commands in `--help` not implemented: push-templates, diff-templates, template-status, symbolic demo, symbolic fragments
+- `status` shows 0 branches — detector may not resolve registry/branch paths
+- Help text is aspirational, not grounded in code reality

 ---

 ## Todos

-
+- Fix or remove unimplemented commands from --help
+- Investigate why status shows 0 branches (registry path resolution)
+- Decide on torch/sentence-transformers — install or provide graceful fallback
@@ -1,14 +1,53 @@
 # TRIGGER Branch-Local Context
-<!-- Source: /home/aipass/aipass_business/AIPass/src/aipass/trigger/.aipass/branch_system_prompt.md -->

-> Auto-created by aipass init. Customize for your branch.
+## Role
+Event bus and error dispatch for AIPass. I detect errors, fingerprint them, gate dispatch, and notify affected branches.

-## Status: NEEDS CONFIGURATION
+## Architecture
+```
+trigger.py (entry point — auto-discovers modules/)
+├── core.py        → Event bus: fire/on/off/status (12 events, 12 handlers)
+├── errors.py      → Error registry CLI: list/detail/suppress/resolve/stats/circuit-breaker
+├── medic.py       → Medic toggle: on/off/status/mute/unmute
+├── log_events.py  → Centralized log watcher (system_logs/)
+└── branch_log_events.py → Branch log watcher (*/logs/*.log)

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+handlers/
+├── error_registry.py  → SHA1 fingerprinting, circuit breaker, exponential backoff
+├── error_reporter.py  → report_error() public API + source fix email pipeline
+├── medic_state.py     → Persistence (trigger_config.json)
+├── log_watcher.py     → Watchdog-based branch log watcher with position tracking
+├── events/            → 12 event handlers (registry.py wires them on first fire)
+└── watchers/          → Centralized log watcher (system_logs/)
+```

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+## Key Commands
+```
+drone @trigger fire <event> [key=val]    # Fire event
+drone @trigger list                      # Show registered handlers
+drone @trigger errors list               # View error registry
+drone @trigger errors stats              # Registry + circuit breaker stats
+drone @trigger medic status              # Medic state
+drone @trigger medic on|off              # Toggle dispatch
+drone @trigger medic mute|unmute @branch # Per-branch control
+drone @trigger branch_log_events status  # Log watcher state
+```
+
+## Dispatch Pipeline (8 gates)
+1. Medic enabled → 2. Branch not muted → 3. Count >= 2 → 4. Not DEV_CENTRAL
+5. Branch in registry → 6. Circuit breaker closed → 7. Per-fingerprint backoff → 8. Rate limit
+
+## Critical Files
+- `trigger_json/trigger_config.json` — medic state, circuit breaker, muted branches
+- `trigger_json/error_registry.json` — all tracked errors
+- `trigger_data.json` — log watcher positions, dedup hashes
+
+## Integration Points
+- **ai_mail**: `deliver_email_to_branch()` for dispatch and source fix emails
+- **prax**: Logger (`from aipass.prax import logger`), prax monitor for live log watching
+- **AIPASS_REGISTRY.json**: Branch validation for dispatch targets
+
+## Rules
+- Never fix errors in other branches — detect and dispatch, they fix their own
+- Hot path logging is poison — event bus fire() must be silent by default
+- Error registry is operational, not archival — clear resolved entries regularly
@@ -41,9 +41,9 @@ def send_source_fix_email(entry: dict) -> bool:
        True if email sent successfully
    """
    try:
-        from aipass.ai_mail.apps.modules.email import send_email_direct
+        from aipass.ai_mail.apps.modules.email import deliver_email_to_branch
    except ImportError:
-        logger.info("[ERRORS] Could not import send_email_direct - ai_mail not available")
+        logger.info("[ERRORS] Could not import deliver_email_to_branch - ai_mail not available")
        return False

    try:
@@ -85,16 +85,20 @@ This will prevent unnecessary error dispatch for this issue.
 Automated recommendation from Medic v2 Error Registry.
 Reply to @trigger with your fix status."""

-        send_email_direct(
-            to_branch=recipient,
-            subject=subject,
-            message=body,
-            reply_to='@trigger',
-            from_branch='@trigger'
-        )
+        from datetime import datetime
+        email_data = {
+            "from": "@trigger",
+            "from_name": "TRIGGER",
+            "to": recipient,
+            "subject": subject,
+            "message": body,
+            "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+        }
+        success, _ = deliver_email_to_branch(recipient, email_data)

-        logger.info(f"[ERRORS] Source fix email sent to {recipient} for {fingerprint}")
-        return True
+        if success:
+            logger.info(f"[ERRORS] Source fix email sent to {recipient} for {fingerprint}")
+        return success
    except Exception as exc:
        logger.info(f"[ERRORS] Failed to send source fix email: {exc}")
        return False
@@ -50,7 +50,7 @@ def _find_repo_root() -> Path:

 _REPO_ROOT = _find_repo_root()

-BRANCH_REGISTRY_FILE = _REPO_ROOT / "BRANCH_REGISTRY.json"
+BRANCH_REGISTRY_FILE = _REPO_ROOT / "AIPASS_REGISTRY.json"
 TRIGGER_CONFIG_FILE = TRIGGER_ROOT / "trigger_json" / "trigger_config.json"

 # Email send callback (set by module layer, avoids handler importing from modules)
@@ -284,7 +284,7 @@ def handle_error_logged(
        # --- Dispatch ---

        try:
-            from aipass.ai_mail.apps.modules.email import send_email_direct
+            from aipass.ai_mail.apps.modules.email import deliver_email_to_branch
        except ImportError:
            return

@@ -301,14 +301,15 @@ def handle_error_logged(
            log_file=effective_log_file
        )

-        send_email_direct(
-            to_branch=recipient,
-            subject=email_subject,
-            message=notification_message,
-            auto_execute=True,
-            reply_to='@trigger',
-            from_branch='@trigger'
-        )
+        email_data = {
+            "from": "@trigger",
+            "from_name": "TRIGGER",
+            "to": recipient,
+            "subject": email_subject,
+            "message": f"⚡ DISPATCH TASK - READ THIS FIRST ⚡\n\n{notification_message}",
+            "timestamp": effective_timestamp,
+        }
+        deliver_email_to_branch(recipient, email_data)

        # Record dispatch for rate limiting
        _record_dispatch(recipient)
@@ -28,8 +28,25 @@ def setup_handlers():

    # Wire up email send callback for error_detected handler (avoids handler importing from modules)
    try:
-        from aipass.ai_mail.apps.modules.email import send_email_direct
-        set_send_email_callback(send_email_direct)
+        from aipass.ai_mail.apps.modules.email import deliver_email_to_branch
+        from datetime import datetime
+
+        def _send_email_adapter(to_branch, subject, message, auto_execute=False, reply_to='@trigger', from_branch='@trigger', **kwargs):
+            """Adapt error_detected handler's call signature to deliver_email_to_branch."""
+            email_data = {
+                "from": from_branch,
+                "from_name": "TRIGGER",
+                "to": to_branch,
+                "subject": subject,
+                "message": message,
+                "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+            }
+            if auto_execute:
+                email_data["message"] = f"⚡ DISPATCH TASK - READ THIS FIRST ⚡\n\n{message}"
+            success, _ = deliver_email_to_branch(to_branch, email_data)
+            return success
+
+        set_send_email_callback(_send_email_adapter)
    except ImportError:
        pass  # ai_mail not available - error notifications won't send
    from .warning_logged import handle_warning_logged
@@ -166,6 +166,7 @@ class Trigger:
    @classmethod
    def status(cls) -> dict:
        """Show registered handlers"""
+        cls._ensure_initialized()
        return {event: len(handlers) for event, handlers in cls._handlers.items()}


@@ -186,6 +187,13 @@ def handle_command(command: str, args: list) -> bool:
    """
    from aipass.cli.apps.modules import console

+    # Handle module-name routing (drone @trigger core <subcmd>)
+    if command == "core":
+        if not args or args[0] in ['--help', '-h', 'help']:
+            _print_help(console)
+            return True
+        return handle_command(args[0], args[1:])
+
    if command not in ["fire", "status", "list"]:
        return False

@@ -127,6 +127,13 @@ def handle_command(command: str, args: list) -> bool:
    """
    from aipass.cli.apps.modules import console

+    # Handle module-name routing (drone @trigger log_events <subcmd>)
+    if command == "log_events":
+        if not args or args[0] in ['--help', '-h', 'help']:
+            print_help()
+            return True
+        return handle_command(args[0], args[1:])
+
    if command not in ["start", "stop", "status"]:
        return False

@@ -1,15 +1,35 @@
 # dev.local.md - TRIGGER
 ```
-Branch: /home/coder/workspace/AIPass/src/aipass/trigger
+Branch: src/aipass/trigger
 Created: 2026-03-07
 ```

 ## Issues

-
+- Centralized log watcher (handlers/watchers/log_watcher.py) is skeleton — not fully wired. Branch log watcher works fine though.
+- Seedgo Diagnostics at 75% — 5 type errors from try/except import pattern ("possibly unbound"). These are intentional graceful degradation, not real bugs.
+- Log watcher systemd service (trigger-log-watcher.service) not set up — medic on/off toggles the flag but can't start/stop the service.

 ---

 ## Todos

-
+- Consider creating the systemd service unit file for persistent log watching
+- Wire up the centralized log watcher fully (system_logs/ watching)
+
+---
+
+## Session Notes
+
+### 2026-03-10 (Session 3) — Full Systems Wiring
+Dispatched by devpulse. Tested everything end-to-end:
+- **Event bus**: 12 events, 12 handlers. fire/list/status all work.
+- **Error registry**: Full lifecycle (report → list → detail → resolve → clear-resolved)
+- **Medic**: on/off/mute/unmute/status all functional
+- **Dispatch pipeline**: Verified 8-gate system. count=1 correctly suppressed, count=2 delivered email to flow inbox.
+- **Bugs fixed**:
+  - `send_email_direct` → `deliver_email_to_branch` (3 files)
+  - `BRANCH_REGISTRY.json` → `AIPASS_REGISTRY.json`
+  - Module-name routing added to core.py, log_events.py
+  - `Trigger.status()` now calls `_ensure_initialized()`
+- **Seedgo**: 98% (22/23 categories at 100%)
@@ -1,14 +1,48 @@
-# {{BRANCHNAME}} Branch-Local Context
-<!-- Source: {{CWD}}/.aipass/branch_system_prompt.md -->
+# COMMONS Branch-Local Context

-> Auto-created by aipass init. Customize for your branch.
+## Role

-## Status: NEEDS CONFIGURATION
+The Commons is the social gathering space for AIPass branches. A community where branches post, comment, vote, browse feeds, join rooms, craft artifacts, explore hidden spaces, and build connections.

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+## Key Commands

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+```bash
+drone @commons post "room" "Title" "Content"   # Post to a room
+drone @commons feed                             # Browse posts
+drone @commons thread <id>                      # View post + comments
+drone @commons comment <id> "text"              # Comment on a post
+drone @commons room list                        # List rooms
+drone @commons enter <room>                     # Enter a room (spatial)
+drone @commons craft "name" "desc"              # Create an artifact
+drone @commons search "query"                   # FTS5 search
+drone @commons who                              # List community members
+drone @commons catchup                          # What you missed
+drone @commons explore                          # Discover secret rooms
+drone @commons --help                           # Full command list
+```
+
+## Architecture
+
+3-layer: Entry point (`apps/commons.py`) -> Modules (`apps/modules/`, 21 thin routers) -> Handlers (`apps/handlers/`, 19 domains). Auto-discovery via `handle_command()`. SQLite with WAL + FTS5. 16 tables.
+
+## Critical Files
+
+- `apps/commons.py` — Entry point, DB init, module discovery
+- `apps/handlers/database/db.py` — Connection manager, schema init
+- `apps/handlers/database/schema.sql` — Flattened schema (16 tables)
+- `apps/handlers/identity/identity_ops.py` — Branch detection via AIPASS_CALLER_CWD
+- `apps/modules/commons_identity.py` — Identity module wrapper
+
+## Key Details
+
+- Commons lives at `src/commons/` (outside `src/aipass/`), so path resolution differs from other branches
+- Branch identity detected via `AIPASS_CALLER_CWD` env var (set by drone) + `.trinity/passport.json` walk-up
+- DB at `~/.aipass/commons.db` (or `$AIPASS_ROOT/.aipass/commons.db`)
+- Registry lookup uses `AIPASS_REGISTRY.json`, found by walking up from package location
+
+## Integration
+
+- All branches can post/comment/vote
+- Branch registration auto-syncs from AIPASS_REGISTRY.json
+- Depends on: `aipass.prax` (logging), `aipass.cli` (console output)
+- Provides: social platform, community feed, artifact system, dashboard data
@@ -24,6 +24,13 @@ import sys
 from pathlib import Path
 from typing import List, Any

+# Fix: When run as a script, Python adds apps/ to sys.path[0] which causes
+# this file (commons.py) to shadow the commons package. Remove it so the
+# installed package resolves correctly.
+_script_dir = str(Path(__file__).resolve().parent)
+if _script_dir in sys.path:
+    sys.path.remove(_script_dir)
+
 # Handle broken pipe gracefully (e.g. output piped to head)
 signal.signal(signal.SIGPIPE, signal.SIG_DFL)

@@ -43,10 +43,10 @@ def _get_db_path() -> Path:
    """
    aipass_root = os.environ.get("AIPASS_ROOT", "")
    if aipass_root:
-        root = Path(aipass_root)
+        root = Path(aipass_root) / ".aipass"
    else:
        root = Path.home() / ".aipass"
-    return root / ".aipass" / "commons.db"
+    return root / "commons.db"


 DB_PATH = _get_db_path()
@@ -303,15 +303,15 @@ def _seed_secret_rooms(conn: sqlite3.Connection) -> None:

 def _register_branches(conn: sqlite3.Connection) -> None:
    """
-    Auto-register all branches from BRANCH_REGISTRY.json as agents.
+    Auto-register all branches from AIPASS_REGISTRY.json as agents.

    Reads the registry and inserts any missing branches. Existing
    branches are left untouched (INSERT OR IGNORE).

-    Searches for BRANCH_REGISTRY.json in standard locations:
+    Searches for AIPASS_REGISTRY.json in standard locations:
    1. AIPASS_ROOT environment variable
-    2. ~/.aipass/BRANCH_REGISTRY.json
-    3. ~/BRANCH_REGISTRY.json (legacy)
+    2. ~/.aipass/AIPASS_REGISTRY.json
+    3. ~/AIPASS_REGISTRY.json (legacy)
    """
    registry_path = _find_branch_registry()
    if not registry_path:
@@ -342,7 +342,7 @@ def _register_branches(conn: sqlite3.Connection) -> None:

 def _find_branch_registry() -> Optional[Path]:
    """
-    Locate BRANCH_REGISTRY.json by searching standard paths.
+    Locate AIPASS_REGISTRY.json by searching standard paths.

    Returns:
        Path to registry file, or None if not found.
@@ -352,12 +352,23 @@ def _find_branch_registry() -> Optional[Path]:
    # Check AIPASS_ROOT env var
    aipass_root = os.environ.get("AIPASS_ROOT", "")
    if aipass_root:
-        search_paths.append(Path(aipass_root) / "BRANCH_REGISTRY.json")
+        search_paths.append(Path(aipass_root) / "AIPASS_REGISTRY.json")
+
+    # Walk up from this package to find project root
+    current = Path(__file__).resolve().parent
+    for _ in range(10):
+        candidate = current / "AIPASS_REGISTRY.json"
+        if candidate.exists():
+            return candidate
+        parent = current.parent
+        if parent == current:
+            break
+        current = parent

    # Standard locations
    search_paths.extend([
-        Path.home() / ".aipass" / "BRANCH_REGISTRY.json",
-        Path.home() / "BRANCH_REGISTRY.json",
+        Path.home() / ".aipass" / "AIPASS_REGISTRY.json",
+        Path.home() / "AIPASS_REGISTRY.json",
    ])

    for path in search_paths:
@@ -14,7 +14,7 @@ caller detection, and mention extraction.

 Detects which branch is calling The Commons based on CWD by walking
 up the directory tree to find a *.id.json file, then cross-referencing
-with BRANCH_REGISTRY.json.
+with AIPASS_REGISTRY.json.
 """

 import os
@@ -32,7 +32,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger

 def _find_branch_registry_path() -> Path:
    """
-    Locate BRANCH_REGISTRY.json by searching standard paths.
+    Locate AIPASS_REGISTRY.json by searching standard paths.

    Returns:
        Path to registry file (may not exist).
@@ -40,20 +40,31 @@ def _find_branch_registry_path() -> Path:
    # Check AIPASS_ROOT env var
    aipass_root = os.environ.get("AIPASS_ROOT", "")
    if aipass_root:
-        candidate = Path(aipass_root) / "BRANCH_REGISTRY.json"
+        candidate = Path(aipass_root) / "AIPASS_REGISTRY.json"
        if candidate.exists():
            return candidate

+    # Walk up from this package to find project root
+    current = Path(__file__).resolve().parent
+    for _ in range(10):
+        candidate = current / "AIPASS_REGISTRY.json"
+        if candidate.exists():
+            return candidate
+        parent = current.parent
+        if parent == current:
+            break
+        current = parent
+
    # Standard locations
    for candidate_path in [
-        Path.home() / ".aipass" / "BRANCH_REGISTRY.json",
-        Path.home() / "BRANCH_REGISTRY.json",
+        Path.home() / ".aipass" / "AIPASS_REGISTRY.json",
+        Path.home() / "AIPASS_REGISTRY.json",
    ]:
        if candidate_path.exists():
            return candidate_path

    # Return a default even if it doesn't exist
-    return Path.home() / "BRANCH_REGISTRY.json"
+    return Path.home() / "AIPASS_REGISTRY.json"


 BRANCH_REGISTRY_PATH = _find_branch_registry_path()
@@ -78,8 +89,8 @@ def find_branch_root(start_path: Path) -> Optional[Path]:
    current = start_path.resolve()

    for _ in range(10):
-        id_files = list(current.glob("*.id.json"))
-        if id_files:
+        # AIPass branches have .trinity/passport.json
+        if (current / ".trinity" / "passport.json").exists():
            return current

        parent = current.parent
@@ -92,7 +103,7 @@ def find_branch_root(start_path: Path) -> Optional[Path]:

 def get_branch_info_from_registry(branch_path: Path) -> Optional[Dict[str, Any]]:
    """
-    Look up branch information in BRANCH_REGISTRY.json by path.
+    Look up branch information in AIPASS_REGISTRY.json by path.

    Args:
        branch_path: Path to branch directory.
@@ -108,9 +119,12 @@ def get_branch_info_from_registry(branch_path: Path) -> Optional[Dict[str, Any]]
            registry = json.load(f)

        branch_path_str = str(branch_path.resolve())
+        registry_root = BRANCH_REGISTRY_PATH.parent

        for branch in registry.get("branches", []):
-            if str(Path(branch["path"]).resolve()) == branch_path_str:
+            # Registry paths are relative to the registry file's parent
+            candidate = (registry_root / branch["path"]).resolve()
+            if str(candidate) == branch_path_str:
                return branch

        return None
@@ -123,7 +137,7 @@ def get_caller_branch() -> Optional[Dict[str, Any]]:
    """
    Detect which branch is calling The Commons based on PWD.

-    Walks up from CWD to find branch root, then looks up in BRANCH_REGISTRY.json.
+    Walks up from CWD to find branch root, then looks up in AIPASS_REGISTRY.json.
    Auto-registers the branch as a Commons agent if not already present.

    Returns:
@@ -131,7 +145,9 @@ def get_caller_branch() -> Optional[Dict[str, Any]]:
        or None if no branch detected.
    """
    try:
-        cwd = Path.cwd()
+        # Use caller's original CWD if routed through drone
+        caller_cwd = os.environ.get("AIPASS_CALLER_CWD", "")
+        cwd = Path(caller_cwd) if caller_cwd else Path.cwd()
        branch_root = find_branch_root(cwd)

        if not branch_root:
@@ -200,7 +216,7 @@ _alias_cache: Optional[Dict[str, str]] = None


 def _load_alias_cache() -> Dict[str, str]:
-    """Load branch alias map from BRANCH_REGISTRY.json (cached)."""
+    """Load branch alias map from AIPASS_REGISTRY.json (cached)."""
    global _alias_cache
    if _alias_cache is not None:
        return _alias_cache
@@ -6,10 +6,22 @@ Created: 2026-03-10

 ## Issues

-
+- 1 type error: `dashboard_writer.py` imports `aipass.devpulse.apps.modules.dashboard` which can't resolve (devpulse dependency)
+- Architecture: missing `logs/`, `dropbox/` dirs (template requirements, low priority)

 ---

 ## Todos

-
+- Investigate devpulse dashboard import — may need conditional import or email devpulse
+
+---
+
+## Session Notes
+
+### 2026-03-10 — First Systems Check
+- **DB init fixed**: 5 bugs — sys.path shadowing, double .aipass path, wrong registry name, registry discovery, identity detection
+- **Seedgo audit**: 98%
+- **All commands working**: feed, post, room, enter, craft, search, explore, leaderboard, capsules, who
+- **15/15 branches registered** as agents
+- **Identity detection**: uses AIPASS_CALLER_CWD + .trinity/passport.json walk-up
@@ -1,14 +1,58 @@
-# {{BRANCHNAME}} Branch-Local Context
-<!-- Source: {{CWD}}/.aipass/branch_system_prompt.md -->
+# SKILLS — Branch Context
+<!-- File: src/skills/.aipass/aipass_local_prompt.md — Injected on every prompt when in skills directory. -->

-> Auto-created by aipass init. Customize for your branch.
+Capability framework for AI agents. Discoverable, validatable, executable skill units across three tiers: markdown-only, with handler, full 3-layer.

-## Status: NEEDS CONFIGURATION
+## Commands

-This file is injected into every AI conversation when working from this branch directory. Configure it with:
+```
+drone @skills list                         # Show all discovered skills
+drone @skills info <name>                  # Display SKILL.md contents
+drone @skills run <name> [action] [args]   # Execute a skill's handler
+drone @skills create <name>                # Scaffold new skill (markdown only)
+drone @skills create <name> --with-handler # Scaffold with handler.py
+drone @skills create <name> --full         # Scaffold with full 3-layer structure
+drone @skills validate <name>              # Check if skill requirements are met
+drone @skills --help                       # Show help
+```

- Who this branch is (role, purpose)
- Key commands and workflows
- Architecture overview
- Critical files and operational rules
- Integration points with other branches
+## Apps Layout
+
+```
+apps/
+├── skills.py              # Entry point — command routing
+├── modules/
+│   ├── discovery.py       # Orchestration: discover_all (thin, delegates to handler)
+│   ├── loader.py          # Orchestration: load_skill (thin, delegates to handler)
+│   ├── runner.py          # Execute skills (handler-based or markdown-only)
+│   ├── creator.py         # Scaffold new skills from templates
+│   └── validator.py       # Check skill requirements
+├── handlers/
+│   ├── discovery_handler.py  # Core: search paths, SKILL.md scanning, frontmatter parsing
+│   ├── loader_handler.py     # Core: parse full SKILL.md, dynamic handler import
+│   ├── registry.py           # Build deduplicated skill registry
+│   ├── validator.py          # Requirement checking (pip, bins, config)
+│   └── template.py           # Template resolution and copying
+├── plugins/               # Extension point (empty)
+catalog/                   # Built-in skills: drone_commands, github, system_status
+templates/                 # Skill creation templates (markdown_only, with_handler, full)
+```
+
+## Search Paths (first match wins)
+
+1. `.aipass/skills/` — Project-local skills
+2. `~/.aipass/skills/` — Global user skills
+3. `src/skills/catalog/` — Built-in skills
+
+## Three Skill Tiers
+
+- **Markdown only**: SKILL.md with instructions (AI reads and follows)
+- **With handler**: SKILL.md + handler.py (programmatic execution)
+- **Full 3-layer**: SKILL.md + apps/ structure (complex skills)
+
+## Memory & Tracking
+
+- `.trinity/passport.json` — identity
+- `.trinity/local.json` — session history
+- `.trinity/observations.json` — collaboration patterns
+- `dev.local.md` — scratchpad for issues, todos, notes
@@ -22,6 +22,7 @@ from pathlib import Path
 from aipass.prax import logger

 # Try yaml, fall back to simple parser
+yaml = None
 try:
    import yaml
    HAS_YAML = True
@@ -6,7 +6,7 @@ Created: 2026-03-10

 ## Issues

-
+- drone routing crash: `Console.print() got unexpected keyword argument 'stderr'` in drone.py:237. Skills commands unreachable via `drone @skills`. Workaround: `python -m skills.apps.skills <cmd>`. Not our bug — lives in drone branch.

 ---

@@ -0,0 +1,106 @@
+# ===================AIPASS====================
+# META DATA HEADER
+# Name: test_cli_routing.py - Unit tests for skills.py CLI routing
+# Date: 2026-03-10
+# Version: 1.0.0
+# Category: skills/tests
+# =============================================
+
+"""Tests for the skills entry point CLI routing."""
+
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+skills_root = Path(__file__).resolve().parent.parent.parent
+if str(skills_root) not in sys.path:
+    sys.path.insert(0, str(skills_root))
+
+from skills.apps.skills import handle_command, _parse_extra_args
+
+
+class TestParseExtraArgs:
+    def test_key_value_pairs(self):
+        result = _parse_extra_args(["host=localhost", "port=8080"])
+        assert result == {"host": "localhost", "port": "8080"}
+
+    def test_positional_args(self):
+        result = _parse_extra_args(["foo", "bar"])
+        assert result == {"arg0": "foo", "arg1": "bar"}
+
+    def test_mixed_args(self):
+        result = _parse_extra_args(["foo", "key=val", "bar"])
+        assert result == {"arg0": "foo", "key": "val", "arg1": "bar"}
+
+    def test_empty_args(self):
+        result = _parse_extra_args([])
+        assert result == {}
+
+    def test_value_with_equals_sign(self):
+        """key=value where value itself contains '='."""
+        result = _parse_extra_args(["query=a=b"])
+        assert result == {"query": "a=b"}
+
+
+class TestHandleCommand:
+    def test_none_command_shows_introspection(self):
+        result = handle_command(None)
+        assert result is True
+
+    def test_help_command(self):
+        result = handle_command("--help")
+        assert result is True
+
+    def test_help_alias(self):
+        result = handle_command("help")
+        assert result is True
+
+    def test_h_flag(self):
+        result = handle_command("-h")
+        assert result is True
+
+    def test_version_command(self):
+        result = handle_command("--version")
+        assert result is True
+
+    def test_version_short_flag(self):
+        result = handle_command("-V")
+        assert result is True
+
+    def test_unknown_command_returns_false(self):
+        result = handle_command("bogus_command_xyz")
+        assert result is False
+
+    def test_list_command(self):
+        result = handle_command("list")
+        assert result is True
+
+    def test_info_missing_args_returns_false(self):
+        result = handle_command("info")
+        assert result is False
+
+    def test_info_with_valid_skill(self):
+        result = handle_command("info", ["github"])
+        assert result is True
+
+    def test_run_missing_args_returns_false(self):
+        result = handle_command("run")
+        assert result is False
+
+    def test_run_with_valid_skill(self):
+        result = handle_command("run", ["system_status", "disk"])
+        assert result is True
+
+    def test_validate_missing_args_returns_false(self):
+        result = handle_command("validate")
+        assert result is False
+
+    def test_validate_with_valid_skill(self):
+        result = handle_command("validate", ["github"])
+        assert result is True
+
+    def test_create_missing_args_returns_false(self):
+        result = handle_command("create")
+        assert result is False
@@ -19,7 +19,7 @@ skills_root = Path(__file__).resolve().parent.parent.parent
 if str(skills_root) not in sys.path:
    sys.path.insert(0, str(skills_root))

-from skills.apps.modules.discovery import (
+from skills.apps.handlers.discovery_handler import (
    _extract_frontmatter,
    _parse_simple_value,
    _simple_frontmatter_parse,
@@ -22,7 +22,7 @@ if str(skills_root) not in sys.path:
 from skills.apps.handlers.template import copy_template, get_template
 from skills.apps.modules.creator import create_skill
 from skills.apps.modules.discovery import discover_skills_in_path, parse_frontmatter
-from skills.apps.modules.loader import _load_handler, _parse_full_skill_md
+from skills.apps.handlers.loader_handler import import_handler, parse_full_skill_md
 from skills.apps.modules.runner import run_skill


@@ -55,7 +55,7 @@ class TestFullLifecycle:
        assert skills[0]["has_handler"] is False

        # Load (parse full SKILL.md)
-        metadata, body = _parse_full_skill_md(skill_path / "SKILL.md")
+        metadata, body = parse_full_skill_md(skill_path / "SKILL.md")
        assert metadata is not None
        assert metadata["name"] == "test-md"
        assert body is not None
@@ -75,7 +75,7 @@ class TestFullLifecycle:
        assert len(handler_skill) == 1

        # Load handler
-        handler = _load_handler(skill_path, "test-handler")
+        handler = import_handler(skill_path, "test-handler")
        assert handler is not None
        assert hasattr(handler, "run")
        assert hasattr(handler, "get_actions")
@@ -113,7 +113,7 @@ class TestCatalogSkillsLifecycle:
        assert github[0]["has_handler"] is False

        # Parse full SKILL.md
-        metadata, body = _parse_full_skill_md(github[0]["path"] / "SKILL.md")
+        metadata, body = parse_full_skill_md(github[0]["path"] / "SKILL.md")
        assert metadata["name"] == "github"
        assert body is not None
        assert "gh" in body.lower()
@@ -56,6 +56,10 @@ class TestLoadSkill:
        metadata = result["metadata"]
        assert "name" in metadata
        assert "description" in metadata
+        # Verify actual values, not just key existence
+        assert metadata["name"] == "github"
+        assert isinstance(metadata["description"], str)
+        assert len(metadata["description"]) > 0

    def test_body_is_markdown_content(self):
        result = load_skill("github")
@@ -0,0 +1,158 @@
+# ===================AIPASS====================
+# META DATA HEADER
+# Name: test_registry.py - Unit tests for skills registry
+# Date: 2026-03-10
+# Version: 1.0.0
+# Category: skills/tests
+# =============================================
+
+"""Tests for the skills registry handler."""
+
+import sys
+import tempfile
+from pathlib import Path
+
+import pytest
+
+skills_root = Path(__file__).resolve().parent.parent.parent
+if str(skills_root) not in sys.path:
+    sys.path.insert(0, str(skills_root))
+
+from skills.apps.handlers.registry import build_registry, get_skill, get_skill_names
+
+
+class TestBuildRegistry:
+    def _make_discover_fn(self, skills_by_path):
+        """Helper: returns a discover_fn that returns skills based on path."""
+        def discover_fn(path, source_label):
+            return skills_by_path.get(str(path), [])
+        return discover_fn
+
+    def test_empty_search_paths(self):
+        registry = build_registry([], lambda p, s: [])
+        assert registry == []
+
+    def test_nonexistent_path_skipped(self):
+        discover_fn = lambda p, s: [{"name": "should-not-appear"}]
+        registry = build_registry(
+            [("/nonexistent/path/xyz_abc_123", "test")],
+            discover_fn,
+        )
+        assert registry == []
+
+    def test_discovers_skills_from_valid_path(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            skill = {"name": "alpha", "description": "A skill", "source": "test"}
+            discover_fn = self._make_discover_fn({tmpdir: [skill]})
+            registry = build_registry([(tmpdir, "test")], discover_fn)
+            assert len(registry) == 1
+            assert registry[0]["name"] == "alpha"
+            assert registry[0]["description"] == "A skill"
+
+    def test_first_match_wins_dedup(self):
+        """When two paths contain a skill with the same name, first path wins."""
+        with tempfile.TemporaryDirectory() as dir1, \
+             tempfile.TemporaryDirectory() as dir2:
+            skill_v1 = {"name": "dupe", "description": "First", "source": "project"}
+            skill_v2 = {"name": "dupe", "description": "Second", "source": "builtin"}
+            discover_fn = self._make_discover_fn({
+                dir1: [skill_v1],
+                dir2: [skill_v2],
+            })
+            registry = build_registry(
+                [(dir1, "project"), (dir2, "builtin")],
+                discover_fn,
+            )
+            assert len(registry) == 1
+            assert registry[0]["description"] == "First"
+            assert registry[0]["source"] == "project"
+
+    def test_different_names_both_included(self):
+        with tempfile.TemporaryDirectory() as dir1, \
+             tempfile.TemporaryDirectory() as dir2:
+            skill_a = {"name": "alpha", "description": "A"}
+            skill_b = {"name": "beta", "description": "B"}
+            discover_fn = self._make_discover_fn({
+                dir1: [skill_a],
+                dir2: [skill_b],
+            })
+            registry = build_registry(
+                [(dir1, "project"), (dir2, "builtin")],
+                discover_fn,
+            )
+            assert len(registry) == 2
+            names = {s["name"] for s in registry}
+            assert names == {"alpha", "beta"}
+
+    def test_multiple_skills_from_single_path(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            skills = [
+                {"name": "one", "description": "First"},
+                {"name": "two", "description": "Second"},
+                {"name": "three", "description": "Third"},
+            ]
+            discover_fn = self._make_discover_fn({tmpdir: skills})
+            registry = build_registry([(tmpdir, "test")], discover_fn)
+            assert len(registry) == 3
+
+    def test_discover_fn_is_called_with_path_and_label(self):
+        """Verify discover_fn receives Path object and source label."""
+        calls = []
+        def tracking_fn(path, source_label):
+            calls.append((path, source_label))
+            return []
+        with tempfile.TemporaryDirectory() as tmpdir:
+            build_registry([(tmpdir, "my_source")], tracking_fn)
+            assert len(calls) == 1
+            assert isinstance(calls[0][0], Path)
+            assert calls[0][1] == "my_source"
+
+
+class TestGetSkill:
+    def test_found(self):
+        registry = [
+            {"name": "alpha", "description": "A"},
+            {"name": "beta", "description": "B"},
+        ]
+        result = get_skill("beta", registry)
+        assert result is not None
+        assert result["name"] == "beta"
+        assert result["description"] == "B"
+
+    def test_not_found(self):
+        registry = [{"name": "alpha", "description": "A"}]
+        result = get_skill("nonexistent", registry)
+        assert result is None
+
+    def test_empty_registry(self):
+        result = get_skill("anything", [])
+        assert result is None
+
+    def test_returns_first_match(self):
+        """If registry somehow has duplicates, returns the first one."""
+        registry = [
+            {"name": "dup", "description": "First"},
+            {"name": "dup", "description": "Second"},
+        ]
+        result = get_skill("dup", registry)
+        assert result["description"] == "First"
+
+
+class TestGetSkillNames:
+    def test_returns_sorted_names(self):
+        registry = [
+            {"name": "charlie"},
+            {"name": "alpha"},
+            {"name": "bravo"},
+        ]
+        names = get_skill_names(registry)
+        assert names == ["alpha", "bravo", "charlie"]
+
+    def test_empty_registry(self):
+        names = get_skill_names([])
+        assert names == []
+
+    def test_single_skill(self):
+        registry = [{"name": "only"}]
+        names = get_skill_names(registry)
+        assert names == ["only"]
@@ -84,15 +84,27 @@ class TestRunSkillReturnContract:
        assert "success" in result
        assert "output" in result
        assert "error" in result
+        # Verify values are correct, not just keys
+        assert result["success"] is True
+        assert "Disk Usage" in result["output"]
+        assert result["error"] is None

    def test_success_result_types(self):
        result = run_skill("system_status", action="disk")
        assert isinstance(result["success"], bool)
        assert isinstance(result["output"], str)
        assert result["error"] is None
+        # Content assertions — not just types
+        assert result["success"] is True
+        assert len(result["output"]) > 0
+        assert "Disk Usage" in result["output"]

    def test_failure_result_types(self):
        result = run_skill("nonexistent_skill_xyz")
        assert isinstance(result["success"], bool)
        assert isinstance(result["output"], str)
        assert isinstance(result["error"], str)
+        # Content assertions — not just types
+        assert result["success"] is False
+        assert "not found" in result["error"].lower()
+        assert result["output"] == ""
@@ -0,0 +1,112 @@
+# ===================AIPASS====================
+# META DATA HEADER
+# Name: test_runner_handler.py - Unit tests for runner_handler (empty body, etc.)
+# Date: 2026-03-10
+# Version: 1.0.0
+# Category: skills/tests
+# =============================================
+
+"""Tests for the skills runner handler, focusing on run_markdown edge cases."""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+skills_root = Path(__file__).resolve().parent.parent.parent
+if str(skills_root) not in sys.path:
+    sys.path.insert(0, str(skills_root))
+
+from skills.apps.handlers.runner_handler import run_markdown, run_handler
+
+
+class TestRunMarkdownEmptyBody:
+    def test_empty_body_returns_success(self):
+        result = run_markdown("empty-skill", {"description": "test"}, "")
+        assert result["success"] is True
+
+    def test_empty_body_output_mentions_no_instructions(self):
+        result = run_markdown("empty-skill", {}, "")
+        assert "no instructions body" in result["output"].lower()
+        assert "empty-skill" in result["output"]
+
+    def test_none_body_returns_no_instructions(self):
+        result = run_markdown("test-skill", {}, None)
+        assert result["success"] is True
+        assert "no instructions body" in result["output"].lower()
+
+    def test_empty_body_no_error(self):
+        result = run_markdown("test-skill", {}, "")
+        assert result["error"] is None
+
+
+class TestRunMarkdownWithBody:
+    def test_body_included_in_output(self):
+        result = run_markdown("my-skill", {"description": "A skill"}, "# Instructions\nDo stuff.")
+        assert result["success"] is True
+        assert "# Instructions" in result["output"]
+        assert "Do stuff." in result["output"]
+
+    def test_header_includes_skill_name(self):
+        result = run_markdown("my-skill", {}, "body content")
+        assert "=== Skill: my-skill ===" in result["output"]
+
+    def test_header_includes_description(self):
+        result = run_markdown("my-skill", {"description": "Does things"}, "body")
+        assert "Does things" in result["output"]
+
+    def test_no_description_still_works(self):
+        result = run_markdown("my-skill", {}, "body")
+        assert result["success"] is True
+        assert "=== Skill: my-skill ===" in result["output"]
+
+
+class TestRunHandler:
+    def test_no_action_with_get_actions(self):
+        """When action is None and handler has get_actions, list them."""
+        class MockHandler:
+            def get_actions(self):
+                return ["disk", "memory"]
+        result = run_handler(MockHandler(), "test-skill", None, {}, {})
+        assert result["success"] is True
+        assert "disk" in result["output"]
+        assert "memory" in result["output"]
+
+    def test_no_action_without_get_actions(self):
+        """When action is None and handler lacks get_actions, return error."""
+        class MockHandler:
+            pass
+        result = run_handler(MockHandler(), "test-skill", None, {}, {})
+        assert result["success"] is False
+        assert "no action specified" in result["error"].lower()
+
+    def test_handler_no_run_function(self):
+        class MockHandler:
+            pass
+        result = run_handler(MockHandler(), "test-skill", "do_stuff", {}, {})
+        assert result["success"] is False
+        assert "no run() function" in result["error"].lower()
+
+    def test_handler_returns_dict(self):
+        class MockHandler:
+            def run(self, action, args=None, config=None):
+                return {"success": True, "output": "done", "error": None}
+        result = run_handler(MockHandler(), "test-skill", "go", {}, {})
+        assert result["success"] is True
+        assert result["output"] == "done"
+
+    def test_handler_returns_non_dict(self):
+        class MockHandler:
+            def run(self, action, args=None, config=None):
+                return "just a string"
+        result = run_handler(MockHandler(), "test-skill", "go", {}, {})
+        assert result["success"] is True
+        assert result["output"] == "just a string"
+
+    def test_handler_raises_exception(self):
+        class MockHandler:
+            def run(self, action, args=None, config=None):
+                raise ValueError("boom")
+        result = run_handler(MockHandler(), "test-skill", "go", {}, {})
+        assert result["success"] is False
+        assert "boom" in result["error"]
@@ -87,3 +87,22 @@ class TestValidateSkill:
        assert "missing_pip" in result
        assert "missing_bins" in result
        assert "missing_config" in result
+        # Verify actual values, not just key existence
+        assert result["valid"] is True
+        assert result["missing_pip"] == []
+        assert result["missing_bins"] == []
+        assert result["missing_config"] == []
+
+    def test_return_structure_with_failures(self):
+        """Verify structure contains actual failure data, not just keys."""
+        result = validate_skill({
+            "requires": {
+                "pip": ["nonexistent_pkg_xyz_123"],
+                "bins": ["nonexistent_bin_xyz"],
+                "config": ["NONEXISTENT_VAR_XYZ"],
+            }
+        })
+        assert result["valid"] is False
+        assert result["missing_pip"] == ["nonexistent_pkg_xyz_123"]
+        assert result["missing_bins"] == ["nonexistent_bin_xyz"]
+        assert result["missing_config"] == ["NONEXISTENT_VAR_XYZ"]