slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

remove: Gemini CLI support — delete .gemini/, GEMINI.md, strip all refs from setup.sh, README, init, prax monitoring, hooks, tests (DPLAN-0188)

Browse Source
This commit is contained in:
AIOSAI
2026-05-24 15:37:48 -07:00
parent 9307cb0ee5
commit 00b1ecff6d
21 changed files with 25 additions and 927 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gemini/hooks/pre_edit_gate.py
-44
View File
@@ -1,44 +0,0 @@
#!/usr/bin/env python3
"""Gemini BeforeTool hook: gate file edits to protect critical files."""
import json
import sys
PROTECTED_PATTERNS = [
".trinity/passport.json",
".aipass/registry.json",
"setup.sh",
]
def main():
try:
input_data = json.loads(sys.stdin.read())
except Exception:
print(json.dumps({}))
return
tool_input = input_data.get("input", {})
file_path = tool_input.get("file_path", "") or tool_input.get("path", "")
if not file_path:
print(json.dumps({}))
return
for pattern in PROTECTED_PATTERNS:
if pattern in file_path:
output = {
"hookSpecificOutput": {
"hookEventName": "BeforeTool",
"permissionDecision": "deny"
},
"systemMessage": f"Edit blocked: {pattern} is a protected file."
}
print(json.dumps(output))
return
print(json.dumps({}))
if __name__ == "__main__":
main()
.gemini/hooks/prompt_inject.py
-97
View File
@@ -1,97 +0,0 @@
#!/usr/bin/env python3
"""Gemini BeforeModel hook: inject per-turn AIPass context."""
import json
import sys
from datetime import datetime
from pathlib import Path
def find_repo_root():
p = Path.cwd()
while p != p.parent:
if (p / ".git").exists():
return p
p = p.parent
return None
def get_branch_from_cwd(repo_root):
cwd = Path.cwd()
try:
rel = cwd.relative_to(repo_root / "src" / "aipass")
return str(rel).split("/")[0]
except ValueError:
try:
rel = cwd.relative_to(repo_root / "src")
return str(rel).split("/")[0]
except ValueError:
return None
def main():
try:
input_data = json.loads(sys.stdin.read())
except Exception:
input_data = {}
repo_root = find_repo_root()
if not repo_root:
print(json.dumps({}))
return
context_parts = []
now = datetime.now().strftime("%A, %B %-d %Y — %-I:%M %p")
context_parts.append(f"# Current Time: {now}")
branch = get_branch_from_cwd(repo_root)
if branch:
branch_dir = repo_root / "src" / "aipass" / branch
if not branch_dir.exists():
branch_dir = repo_root / "src" / branch
passport = branch_dir / ".trinity" / "passport.json"
if passport.exists():
try:
data = json.loads(passport.read_text(encoding="utf-8"))
identity = data.get("identity", {})
traits = data.get("traits", [])
context_parts.append(
f"# {branch.upper()} Identity\n"
f"Path: {data.get('branch_info', {}).get('path', 'unknown')}\n"
f"Role: {identity.get('role', 'unknown')}\n"
f"Traits: {' | '.join(traits)}\n"
f"Purpose: {identity.get('purpose', 'unknown')}"
)
except Exception:
pass
inbox = branch_dir / ".ai_mail.local" / "inbox.json"
if inbox.exists():
try:
mail = json.loads(inbox.read_text(encoding="utf-8"))
unread = mail.get("unread_count", 0)
if unread > 0:
context_parts.append(
f"You have {unread} new emails - check with: "
f"drone @ai_mail inbox"
)
except Exception:
pass
if context_parts:
context = "\n\n".join(context_parts)
output = {
"hookSpecificOutput": {
"hookEventName": "BeforeModel",
"additionalContext": context
}
}
else:
output = {}
print(json.dumps(output))
if __name__ == "__main__":
main()
.gemini/hooks/session_start_identity.py
-86
View File
@@ -1,86 +0,0 @@
#!/usr/bin/env python3
"""Gemini SessionStart hook: inject AIPass identity context."""
import json
import sys
from pathlib import Path
def find_repo_root():
p = Path.cwd()
while p != p.parent:
if (p / ".git").exists():
return p
p = p.parent
return None
def get_branch_from_cwd(repo_root):
cwd = Path.cwd()
try:
rel = cwd.relative_to(repo_root / "src" / "aipass")
return str(rel).split("/")[0]
except ValueError:
try:
rel = cwd.relative_to(repo_root / "src")
return str(rel).split("/")[0]
except ValueError:
return None
def main():
try:
input_data = json.loads(sys.stdin.read())
except Exception:
input_data = {}
repo_root = find_repo_root()
if not repo_root:
print(json.dumps({}))
return
context_parts = []
global_prompt = repo_root / ".aipass" / "aipass_global_prompt.md"
if global_prompt.exists():
context_parts.append(global_prompt.read_text(encoding="utf-8")[:8000])
branch = get_branch_from_cwd(repo_root)
if branch:
branch_dir = repo_root / "src" / "aipass" / branch
if not branch_dir.exists():
branch_dir = repo_root / "src" / branch
passport = branch_dir / ".trinity" / "passport.json"
if passport.exists():
try:
data = json.loads(passport.read_text(encoding="utf-8"))
identity = data.get("identity", {})
context_parts.append(
f"# Branch Identity: {branch.upper()}\n"
f"Role: {identity.get('role', 'unknown')}\n"
f"Purpose: {identity.get('purpose', 'unknown')}\n"
f"Class: {identity.get('citizen_class', 'unknown')}"
)
except Exception:
pass
branch_prompt = branch_dir / ".aipass" / "aipass_local_prompt.md"
if branch_prompt.exists():
context_parts.append(branch_prompt.read_text(encoding="utf-8")[:4000])
if context_parts:
context = "\n\n---\n\n".join(context_parts)
output = {
"hookSpecificOutput": {
"hookEventName": "SessionStart",
"additionalContext": context
}
}
else:
output = {}
print(json.dumps(output))
if __name__ == "__main__":
main()
.gemini/skills/memo/SKILL.md
-28
View File
@@ -1,28 +0,0 @@
---
name: memo
description: Update branch memory files after completing work. Saves session history, key learnings, and collaboration observations to .trinity/ files.
---
# Memory Update
Purpose: Update branch memory files after completing work this session.
## Execution
1. Read `.trinity/passport.json` first — re-absorb your identity, role, and principles before writing memories
2. Review what was done this session (context, recent changes, key decisions)
3. Update each file below as needed
4. Confirm completion — list files updated
## What to Update
### Always
- **.trinity/local.json** — Add new session entry to `sessions` if significant work was done. Add new `key_learnings` for facts you'd need next time. Trim oldest sessions if over 20.
- **.trinity/observations.json** — Add notable collaboration insights: breakthrough moments, pattern corrections, flow states, friction points, preference discoveries. Skip if nothing notable this session.
### If Relevant
- **.trinity/passport.json** — Evolve identity when the branch's role, capabilities, or principles have genuinely changed. Don't update just to update — but don't leave placeholders forever either.
- **README.md** — Does it reflect current state? Update if stale.
- **STATUS.local.md** — Drop quick notes on issues, todos, or ideas in the Notepad section.
.gemini/skills/prep/SKILL.md
-56
View File
@@ -1,56 +0,0 @@
---
name: prep
description: Session wrap-up. Update memories, check plans, review git state, check inbox, flag loose ends. Use before closing a session or compacting context.
---
# Session Wrap-Up
Purpose: Button up everything at the end of a session — or before context compaction. Memories, plans, git — all tidy.
## Execution
1. Read `.trinity/passport.json` first — re-absorb your identity before writing anything
2. Do ALL of the following, then confirm what was updated
## 1. Memories
- **.trinity/local.json** — Add/update session entry with summary of work done. Add new key_learnings for anything learned this session. Trim oldest sessions if over 20.
- **.trinity/observations.json** — Add collaboration insights if anything notable happened. Skip if nothing new.
- **.trinity/passport.json** — Only update if role/purpose/principles genuinely changed this session.
## 2. Active Plans
- Check any DPLANs or FPLANs referenced in this session
- Update their execution logs, status, decision logs with current state
- If a plan was completed, note it (but don't close — the user does that)
## 3. Git State
- Run `git status` — report uncommitted changes
- If there's a logical commit waiting, suggest it (don't commit without asking)
- Note the current branch and any open PRs
## 4. Inbox
- Run `drone @ai_mail inbox 2>/dev/null` — report any unread emails
- Close any that were already processed but not formally closed
## 5. Loose Ends
- Flag anything in-flight: running background agents, dispatched branches waiting for replies, pending decisions
- If anything can't survive compaction, write it to STATUS.local.md Notepad
## Confirm
List everything updated. Format:
```
Prep complete:
- local.json: [what was added]
- observations.json: [updated / skipped]
- Plans: [which ones updated]
- Git: [branch, uncommitted count, suggestion]
- Inbox: [count, action taken]
- Loose ends: [any flagged]
Ready to close out or compact.
```
.github/ISSUE_TEMPLATE/bug-report.yml
+1 -1
View File
@@ -35,7 +35,7 @@ body:
placeholder: |
- OS: Ubuntu 24.04
- Python: 3.12
- CLI: Claude Code / Codex / Gemini
- CLI: Claude Code / Codex
validations:
required: true
GEMINI.md
-19
View File
@@ -1,19 +0,0 @@
# AIPass
Multi-agent framework. Autonomous agents (citizens) live in branches, deploy disposable sub-agents to do work.
User: user
# Startup protocol
On any greeting, silently read these files from CWD and run the commands — no narration, no announcing steps. Just do it and respond with the status.
- Read: `.trinity/passport.json`, `.trinity/local.json`, `.trinity/observations.json`, `README.md`, `STATUS.local.md`
- Check: `drone @ai_mail inbox` — process any mail, don't ask.
- Run: `drone @git status`
Use drone commands for all operations. Never raw git, gh, file access, or python -m when drone provides it.
# Memories
Update `.trinity/` at natural breakpoints, after milestones, and on `/memo`.
README.md
+4 -4
View File
@@ -71,7 +71,7 @@ aipass init agent my-agent # Full agent: apps, mail, memory, identity
- **Shared workspace.** All agents work on the same filesystem, same project, same time. No sandboxes.
- **One command for everything.** AIPass ships with `drone`, a CLI router — `drone @agent command` reaches any agent. Learn it once, use it everywhere.
**Runs on your existing CLI subscription.** Claude Pro/Max, Codex, or Gemini — AIPass uses the same CLI binary you already run. No extra API keys, no extra costs for core functionality.
**Runs on your existing CLI subscription.** Claude Pro/Max or Codex — AIPass uses the same CLI binary you already run. No extra API keys, no extra costs for core functionality.
---
@@ -263,7 +263,7 @@ AIPass stores everything locally in your project directory. To remove it:
```bash
# Remove AIPass files from your project
rm -rf .aipass/ .claude/ .ai_mail.local/ hooks/ src/
rm -f CLAUDE.md AGENTS.md GEMINI.md STATUS.local.md *_REGISTRY.json .gitignore
rm -f CLAUDE.md AGENTS.md STATUS.local.md *_REGISTRY.json .gitignore
# If you installed via pip
pip uninstall aipass
@@ -288,7 +288,7 @@ This archives the agent's directory and removes it from the registry.
### Use your existing subscription
AIPass runs on your **existing CLI subscription** — Claude Pro/Max, Codex, or Gemini. No API keys required for core functionality. No extra costs beyond your existing subscription.
AIPass runs on your **existing CLI subscription** — Claude Pro/Max or Codex. No API keys required for core functionality. No extra costs beyond your existing subscription.
This works because AIPass runs each CLI as an **official subprocess** — the same binary you'd run yourself in a terminal. It doesn't extract credentials, proxy API calls, or intercept tokens. Your subscription stays within the provider's infrastructure at all times.
@@ -299,7 +299,7 @@ This works because AIPass runs each CLI as an **official subprocess** — the sa
- Bypass rate limits or prompt caching
- Impersonate official CLI clients
Claude Code is proprietary but officially supports hooks and subprocess usage. Codex and Gemini CLI are open source (Apache 2.0).
Claude Code is proprietary but officially supports hooks and subprocess usage. Codex CLI is open source (Apache 2.0).
> API keys are only needed for optional add-on agents (OpenRouter/OpenAI). For server/automated deployments, API key authentication is recommended per [Anthropic's guidance](https://code.claude.com/docs/en/legal-and-compliance).
setup.sh
+1 -54
View File
@@ -499,7 +499,7 @@ fi
# --- Install Claude Code hooks ---
CLAUDE_SETTINGS="$HOME/.claude/settings.json"
# Determine python command for non-Claude provider hooks (Gemini, etc).
# Determine python command for non-Claude provider hooks.
# Claude hooks use bridge pattern with $AIPASS_HOME env var — no HOOK_PYTHON needed.
# Linux: keep "python3" — distros ship 3.10+ and hooks import nothing
# version-specific beyond that.
@@ -713,58 +713,6 @@ else
echo "Skipping Codex CLI (not installed)"
fi
# --- Install Gemini CLI hooks ---
if command -v gemini &>/dev/null; then
if [ -d "$SCRIPT_DIR/.gemini/hooks" ]; then
echo "Installing Gemini CLI hooks ..."
GEMINI_SETTINGS="$HOME/.gemini/settings.json"
mkdir -p "$HOME/.gemini"
# HOOK_PYTHON was set earlier in the Claude hooks block; reuse it.
# Fall back to python3 if this block runs without that setup (defensive).
GEMINI_HOOK_PYTHON="${HOOK_PYTHON:-python3}"
python3 - "$SCRIPT_DIR" "$GEMINI_SETTINGS" "$GEMINI_HOOK_PYTHON" << 'PYEOF'
import json
import sys
from pathlib import Path
repo_root = sys.argv[1]
settings_path = Path(sys.argv[2])
hook_python = sys.argv[3]
hooks_dir = f"{repo_root}/.gemini/hooks"
# Load existing settings or start fresh
if settings_path.exists():
settings = json.loads(settings_path.read_text())
else:
settings = {}
# Build hooks config with absolute paths (Gemini uses different event names)
settings["hooks"] = {
"SessionStart": [
{"hooks": [{"type": "command", "command": f"{hook_python} {hooks_dir}/session_start_identity.py", "timeout": 10}]}
],
"BeforeModel": [
{"hooks": [{"type": "command", "command": f"{hook_python} {hooks_dir}/prompt_inject.py", "timeout": 10}]}
],
"BeforeTool": [
{"matcher": "Edit|Write",
"hooks": [{"type": "command", "command": f"{hook_python} {hooks_dir}/pre_edit_gate.py", "timeout": 5}]}
],
}
settings_path.write_text(json.dumps(settings, indent=2) + "\n")
print(f" hooks -> {settings_path}")
PYEOF
else
echo "Skipping Gemini hooks (no .gemini/hooks/ directory found in repo)"
fi
else
echo "Skipping Gemini CLI (not installed)"
fi
# --- Set AIPASS_HOME + PATH so all services work from any project ---
echo ""
echo "Configuring cross-project access ..."
@@ -944,7 +892,6 @@ if [ "$FAIL" -eq 0 ]; then
echo "CLI integrations:"
echo " Claude Code: hooks installed to ~/.claude/settings.json"
command -v codex &>/dev/null && echo " Codex CLI: hooks at .codex/hooks.json + config at ~/.codex/config.toml"
command -v gemini &>/dev/null && echo " Gemini CLI: hooks installed to ~/.gemini/settings.json"
echo ""
else
echo "=== Setup finished with errors ==="
src/aipass/aipass/.aipass/aipass_local_prompt.md
+1 -1
View File
@@ -65,7 +65,7 @@ apps/
## Integration
- **Depends on:** @drone (routing), @seedgo (audit), @spawn (first agent creation), @flow (plan test open/close), @ai_mail (test emails), @prax (health signals), pytest, CLI tools (Claude/Codex/Gemini)
- **Depends on:** @drone (routing), @seedgo (audit), @spawn (first agent creation), @flow (plan test open/close), @ai_mail (test emails), @prax (health signals), pytest, CLI tools (Claude/Codex)
- **Serves:** New users first. Also humans asking "how does this work?" anywhere ecosystem.
- **Nothing depends on me.** One-way relationship. Can be removed/replaced without ripple.
src/aipass/aipass/apps/handlers/init/bootstrap.py
+6 -7
View File
@@ -14,13 +14,12 @@ Business logic for `aipass init`. Creates the project scaffold:
2. .aipass/aipass_global_prompt.md — global prompt (injected every turn)
3. CLAUDE.md — project prompt (Claude Code reads this)
4. AGENTS.md — Codex equivalent of CLAUDE.md
5. GEMINI.md — Gemini equivalent of CLAUDE.md
6. README.md — getting started guide
7. STATUS.local.md — project status
8. .gitignore — standard AIPass ignores
9. .claude/settings.json — Claude Code hooks configuration
10. src/ — directory where agents live
11. .ai_mail.local/inbox.json — empty project mailbox
5. README.md — getting started guide
6. STATUS.local.md — project status
7. .gitignore — standard AIPass ignores
8. .claude/settings.json — Claude Code hooks configuration
9. src/ — directory where agents live
10. .ai_mail.local/inbox.json — empty project mailbox
Projects are NOT citizens — no .trinity/ directory. Identity lives in the
registry JSON. Init is re-runnable: existing files are skipped, not errors.
src/aipass/aipass/apps/handlers/init/scaffold_content.py
-1
View File
@@ -52,7 +52,6 @@ def readme_md(name: str) -> str:
" .aipass/ # Prompts (injected per-turn)\n"
" CLAUDE.md # Claude Code instructions\n"
" AGENTS.md # Codex instructions\n"
" GEMINI.md # Gemini instructions\n"
" STATUS.local.md # Project status\n"
" src/ # Agent directories live here\n"
" <agent_name>/ # Created via aipass init agent\n"
src/aipass/aipass/tests/test_handoff_platform.py
+1 -1
View File
@@ -50,7 +50,7 @@ class TestBuildCliCmd:
def test_other_cli(self) -> None:
"""Other CLI names are returned as-is with default variant."""
assert build_cli_cmd("gemini", "default") == "gemini"
assert build_cli_cmd("codex", "default") == "codex"
# =============================================================================
src/aipass/hooks/.aipass/aipass_local_prompt.md
+2 -2
View File
@@ -4,13 +4,13 @@ Injected every turn. Breadcrumbs only -- details in README, --help, .trinity/, S
## Identity
HOOKS -- hook infrastructure owner. Single engine dispatches all hooks across platforms (Claude, Codex, Gemini) with per-project config, full logging, and crash isolation. Builder citizen. The 13th citizen.
HOOKS -- hook infrastructure owner. Single engine dispatches all hooks across platforms (Claude, Codex) with per-project config, full logging, and crash isolation. Builder citizen. The 13th citizen.
## What I Do
- Own the hook engine -- receives events from platform bridges, routes to handlers, logs everything
- Maintain 14 native handlers across 4 categories (prompt, security, lifecycle, notification)
- Bridge platforms -- thin normalization layer per provider (Claude today, Codex/Gemini planned)
- Bridge platforms -- thin normalization layer per provider (Claude today, Codex planned)
- Per-project config -- `.aipass/hooks.json` controls what fires per project
- Log everything -- prax integration + JSONL diagnostics for every hook execution
src/aipass/hooks/README.md
+1 -1
View File
@@ -2,7 +2,7 @@
# Hooks
> Hook infrastructure for AIPass. Single engine dispatches all hooks across platforms (Claude, Codex, Gemini) with per-project config, full logging, and testability. The 13th citizen.
> Hook infrastructure for AIPass. Single engine dispatches all hooks across platforms (Claude, Codex) with per-project config, full logging, and testability. The 13th citizen.
Every hook event flows through one engine. Platform bridges normalize the event format, the engine reads per-project config (.aipass/hooks.json), dispatches matching handlers, and logs everything to prax + JSONL.
src/aipass/prax/README.md
+3 -3
View File
@@ -50,7 +50,7 @@ Real-time unified console showing:
- File changes, log events, drone commands, agent activity
- **Caller attribution** — `CALLER → TARGET` for drone commands
- **Model tags** — `[BRANCH/model]` (e.g., `[DEVPULSE/opus]`, `[DEVPULSE/gpt-5.4]`)
- **Multi-CLI** — Claude Code (JSONL), Codex (JSONL), Gemini (JSON) session monitoring
- **Multi-CLI** — Claude Code (JSONL), Codex (JSONL) session monitoring
- **Polling fallback** — automatic fallback when inotify watches are exhausted
- **Soft start** — only shows new activity after launch (seeks to EOF on startup)
@@ -165,7 +165,7 @@ drone @prax monitor run
2. **Two-tier logging** — Each log entry goes to both `system_logs/` (central, all branches) and `<branch>/logs/` (branch-local), both with size-based rotation.
3. **Self-healing** — Auto-creates missing log directories, falls back to `system_logs/external/` for unknown modules, provides NullLogger if prax itself fails to import.
4. **Mission Control** — Three threads: display worker (pulls from event queue), file watcher (watchdog on branch `apps/` dirs), log watcher (tails `system_logs/*.log`). Falls back to polling when inotify is exhausted.
5. **Multi-CLI monitoring** — Watches Claude Code JSONL, Codex JSONL, and Gemini JSON session files. Extracts agent activity (thinking, tool use, responses) with model detection and branch resolution.
5. **Multi-CLI monitoring** — Watches Claude Code JSONL and Codex JSONL session files. Extracts agent activity (thinking, tool use, responses) with model detection and branch resolution.
6. **Dashboard** — Template-based per-branch dashboard files. Refreshes from central files (`*.central.json`). Write-through API for services to update sections directly.
7. **STATUS sync** — Scans all branch `STATUS.local.md` files, extracts State/Last update fields, builds aggregated `STATUS.md` at the repo root.
@@ -175,7 +175,7 @@ drone @prax monitor run
| Test File | Tests | Coverage |
|-----------|-------|----------|
| test_filesystem_handler.py | 172 | Multi-CLI adapters, Codex/Gemini branch detection |
| test_filesystem_handler.py | 172 | Multi-CLI adapters, Codex branch detection |
| test_monitoring_handlers.py | 139 | Branch detector, stream output, event handling |
| test_operations.py | 99 | Dashboard operations, write-through |
| test_log_watcher.py | 82 | Log file tailing, agent activity parsing |
src/aipass/prax/apps/handlers/monitoring/filesystem_handler.py
+3 -141
View File
@@ -12,7 +12,7 @@ PRAX Filesystem Event Handler
Watchdog FileSystemEventHandler that processes filesystem events and:
1. Pushes events to the monitoring event queue for display
2. Fires trigger events for cross-module integration
3. Parses CLI session files for agent activity tracking (Claude Code, Codex, Gemini)
3. Parses CLI session files for agent activity tracking (Claude Code, Codex)
Extracted from monitor.py to maintain 3-layer architecture:
module (orchestration) -> handler (implementation)
@@ -25,15 +25,13 @@ from typing import Optional, Dict
from watchdog.events import FileSystemEventHandler
from aipass.prax.apps.modules.logger import get_direct_logger
logger = get_direct_logger()
# Monitoring subsystem imports
from aipass.prax.apps.handlers.monitoring.event_queue import MonitoringEvent, MonitoringQueue
from aipass.prax.apps.handlers.monitoring.branch_detector import detect_branch_from_path
from aipass.prax.apps.handlers.monitoring.monitoring_filters import should_monitor, get_priority
from aipass.prax.apps.handlers.json import json_handler
logger = get_direct_logger()
class MonitoringFileHandler(FileSystemEventHandler):
"""
@@ -241,42 +239,6 @@ class MonitoringFileHandler(FileSystemEventHandler):
self._session_branches[path_key] = result
return result
def _resolve_gemini_slug(self, slug: str) -> Optional[str]:
"""Resolve a Gemini project slug to a branch name via projects.json."""
projects_file = Path.home() / ".gemini" / "projects.json"
if not projects_file.exists():
return None
try:
data = _json.loads(projects_file.read_text())
except (OSError, _json.JSONDecodeError) as e:
logger.info(f"[monitor] Could not read Gemini projects.json: {e}")
return None
for project_path, project_slug in data.get("projects", {}).items():
if project_slug == slug:
return self._branch_from_cwd(project_path)
return None
def _get_gemini_branch(self, file_path, path_key: str) -> str:
"""Get branch for a Gemini session from project slug in path."""
if path_key in self._session_branches:
return self._session_branches[path_key]
# Path: ~/.gemini/tmp/<project-slug>/chats/session-*.json
parts = Path(file_path).parts
slug = None
if "tmp" in parts:
idx = parts.index("tmp")
if idx + 1 < len(parts):
slug = parts[idx + 1]
if not slug:
self._session_branches[path_key] = "GEMINI"
return "GEMINI"
branch = self._resolve_gemini_slug(slug) or slug.upper()
self._session_branches[path_key] = branch
return branch
# =========================================================================
# MODEL TAG HELPERS
# =========================================================================
@@ -299,11 +261,6 @@ class MonitoringFileHandler(FileSystemEventHandler):
return m # already short: gpt-4o, gpt-5.4
if m.startswith("o") and any(c.isdigit() for c in m):
return m # o1, o3, o4-mini
# Gemini models
if "gemini" in m:
# gemini-3-flash-preview → gemini-3-flash
parts = m.replace("gemini-", "").split("-")
return "gemini-" + "-".join(p for p in parts if p != "preview")
# Fallback: first 15 chars
return model[:15]
@@ -368,95 +325,6 @@ class MonitoringFileHandler(FileSystemEventHandler):
return f"💬 {part['text'][:120]}"
return "💬 Agent response"
# =========================================================================
# GEMINI AGENT ACTIVITY PARSING (full JSON sessions)
# =========================================================================
@staticmethod
def _extract_gemini_action(message: dict) -> Optional[str]:
"""Extract a display action string from a Gemini session message."""
msg_type = message.get("type", "")
if msg_type == "user":
return "📩 User message"
if msg_type != "gemini":
return None
tool_calls = message.get("toolCalls", [])
if tool_calls:
last_tool = tool_calls[-1]
return f"🔧 {last_tool.get('displayName', last_tool.get('name', 'tool'))}"
if message.get("thoughts"):
return "💭 Thinking"
return MonitoringFileHandler._extract_text_from_content(message.get("content", []))
@staticmethod
def _extract_text_from_content(content) -> str:
"""Extract first text snippet from a content field (list or string)."""
if isinstance(content, str) and content.strip():
return f"💬 {content.strip()[:120]}"
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get("text", "").strip():
return f"💬 {part['text'].strip()[:120]}"
return "💬 Agent response"
def _parse_gemini_activity(self, file_path, branch):
"""Parse Gemini session JSON to show agent actions.
Gemini rewrites the entire file on each change, so we track
the message count and only process new messages.
"""
try:
path_key = str(file_path)
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
data = _json.load(f)
messages = data.get("messages", [])
last_count = self._jsonl_positions.get(path_key, 0)
if len(messages) <= last_count:
return True # No new messages
self._jsonl_positions[path_key] = len(messages)
# Extract model from gemini messages
for msg in messages:
model = msg.get("model", "")
if model:
self._session_models[path_key] = self._shorten_model(model)
break
# Process new messages (most recent first for dedup)
new_messages = messages[last_count:]
for msg in reversed(new_messages):
action_text = self._extract_gemini_action(msg)
if not action_text:
continue
if self._last_agent_action.get(path_key) == action_text:
return True
self._last_agent_action[path_key] = action_text
tagged_branch = self._tag_branch_with_model(path_key, branch)
evt = MonitoringEvent(
priority=1,
event_type="agent",
branch=tagged_branch,
action="activity",
message=action_text,
level="info",
)
if self._event_queue:
self._event_queue.enqueue(evt)
return True
return True
except (_json.JSONDecodeError, OSError) as e:
logger.info(f"[monitor] Gemini parse error for {file_path.name}: {e}")
return False
def _read_new_jsonl_lines(self, file_path, path_key: str) -> Optional[list]:
"""Read new lines from a JSONL file since last position. Returns None if no new data."""
current_size = file_path.stat().st_size
@@ -656,12 +524,6 @@ class MonitoringFileHandler(FileSystemEventHandler):
if self._parse_codex_activity(file_path, codex_branch):
return
# Gemini JSON session files: parse agent activity
if file_path.suffix == ".json" and ".gemini/tmp/" in path_str and "/chats/" in path_str:
gemini_branch = self._get_gemini_branch(file_path, path_str)
if self._parse_gemini_activity(file_path, gemini_branch):
return
self._check_command_indicator(action, file_path, branch)
priority_level = get_priority(file_path, action)
src/aipass/prax/apps/handlers/monitoring/monitoring_filters.py
+1 -4
View File
@@ -72,7 +72,6 @@ MONITOR_IGNORE_PATTERNS = [
".var",
".backup",
".antigravity",
".gemini",
# IDE and editor directories (auto-generated, large)
".vscode/cli",
".vscode/extensions",
@@ -280,11 +279,9 @@ def should_monitor(path: Path) -> bool:
if ".claude.json.backup" in name or ".claude.json.tmp" in name:
return False
# CLI session files: always monitor (override ignore patterns for .codex, .gemini)
# CLI session files: always monitor (override ignore patterns for .codex)
if ".codex/sessions/" in path_str and name.endswith(".jsonl"):
return True
if ".gemini/tmp/" in path_str and "/chats/" in path_str and name.endswith(".json"):
return True
if _matches_always_patterns(path_str, parts, name):
return True
src/aipass/prax/apps/modules/monitor.py
-4
View File
@@ -398,10 +398,6 @@ def _get_watch_directories(repo_root: Path) -> list[tuple[Path, bool]]:
if codex_sessions.exists():
dirs.append((codex_sessions, True))
gemini_tmp = Path.home() / ".gemini" / "tmp"
if gemini_tmp.exists():
dirs.append((gemini_tmp, True))
return dirs
src/aipass/prax/tests/test_filesystem_handler.py
+1 -360
View File
@@ -9,7 +9,7 @@
"""Unit tests for MonitoringFileHandler.
Covers: __init__, watchdog event methods, agent activity parsing
(Claude Code, Codex, Gemini), branch detection, model tag helpers,
(Claude Code, Codex), branch detection, model tag helpers,
internal event processing, and display name building.
"""
@@ -737,105 +737,6 @@ class TestGetCodexBranch:
assert result == "CODEX"
# =============================================
# RESOLVE GEMINI SLUG TESTS
# =============================================
class TestResolveGeminiSlug:
"""Tests for _resolve_gemini_slug."""
def test_resolves_slug_to_branch(self):
"""Should resolve slug to branch via projects.json."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
projects_data = json.dumps({"projects": {"/home/user/Projects/AIPass/src/aipass/prax": "my-slug"}})
with patch.object(Path, "exists", return_value=True):
with patch.object(Path, "read_text", return_value=projects_data):
result = handler._resolve_gemini_slug("my-slug")
assert result == "PRAX"
def test_no_projects_file(self):
"""Should return None when projects.json does not exist."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
with patch.object(Path, "exists", return_value=False):
result = handler._resolve_gemini_slug("my-slug")
assert result is None
def test_os_error_reading_projects(self):
"""Should return None on OSError reading projects.json."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
with patch.object(Path, "exists", return_value=True):
with patch.object(Path, "read_text", side_effect=OSError("fail")):
result = handler._resolve_gemini_slug("my-slug")
assert result is None
def test_json_error_reading_projects(self):
"""Should return None on JSON decode error."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
with patch.object(Path, "exists", return_value=True):
with patch.object(Path, "read_text", return_value="not json"):
result = handler._resolve_gemini_slug("my-slug")
assert result is None
def test_slug_not_found(self):
"""Should return None when slug is not in projects.json."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
projects_data = json.dumps({"projects": {"/some/path": "other-slug"}})
with patch.object(Path, "exists", return_value=True):
with patch.object(Path, "read_text", return_value=projects_data):
result = handler._resolve_gemini_slug("my-slug")
assert result is None
# =============================================
# GET GEMINI BRANCH TESTS
# =============================================
class TestGetGeminiBranch:
"""Tests for _get_gemini_branch."""
def test_cached_branch(self):
"""Should return cached branch if present."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
handler._session_branches["gkey"] = "CACHED"
result = handler._get_gemini_branch("/fake/file.json", "gkey")
assert result == "CACHED"
def test_extracts_slug_from_path(self):
"""Should extract slug from path and resolve it."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
file_path = "/home/user/.gemini/tmp/my-project/chats/session-1.json"
with patch.object(handler, "_resolve_gemini_slug", return_value="PRAX"):
result = handler._get_gemini_branch(file_path, "gkey2")
assert result == "PRAX"
assert handler._session_branches["gkey2"] == "PRAX"
def test_no_slug_returns_gemini(self):
"""Should return GEMINI when no slug can be extracted."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
file_path = "/home/user/.gemini/chats/session-1.json"
result = handler._get_gemini_branch(file_path, "gkey3")
assert result == "GEMINI"
def test_unresolved_slug_uses_uppercase(self):
"""Should use slug.upper() when resolution returns None."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod)
file_path = "/home/user/.gemini/tmp/my-project/chats/session-1.json"
with patch.object(handler, "_resolve_gemini_slug", return_value=None):
result = handler._get_gemini_branch(file_path, "gkey4")
assert result == "MY-PROJECT"
# =============================================
# SHORTEN MODEL TESTS
# =============================================
@@ -878,12 +779,6 @@ class TestShortenModel:
result = mod.MonitoringFileHandler._shorten_model("o4-mini")
assert result == "o4-mini"
def test_gemini_model(self):
"""Should shorten gemini models, removing preview suffix."""
mod, _, _, _, _ = _import_filesystem_handler()
result = mod.MonitoringFileHandler._shorten_model("gemini-3-flash-preview")
assert result == "gemini-3-flash"
def test_fallback_long_model(self):
"""Should truncate unknown models to 15 chars."""
mod, _, _, _, _ = _import_filesystem_handler()
@@ -1116,242 +1011,6 @@ class TestParseCodexResponseItem:
assert "tool" in result
# =============================================
# EXTRACT GEMINI ACTION TESTS
# =============================================
class TestExtractGeminiAction:
"""Tests for _extract_gemini_action static method."""
def test_user_message(self):
"""User messages should return user message."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_gemini_action({"type": "user"})
assert "User message" in result
def test_non_gemini_type_returns_none(self):
"""Non-user, non-gemini types should return None."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_gemini_action({"type": "system"})
assert result is None
def test_gemini_with_tool_calls(self):
"""Gemini message with tool calls should format last tool."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
msg = {
"type": "gemini",
"toolCalls": [
{"displayName": "first_tool"},
{"displayName": "last_tool"},
],
}
result = cls._extract_gemini_action(msg)
assert "last_tool" in result
def test_gemini_with_tool_calls_name_fallback(self):
"""Gemini tool call should fall back to name if no displayName."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
msg = {"type": "gemini", "toolCalls": [{"name": "my_tool"}]}
result = cls._extract_gemini_action(msg)
assert "my_tool" in result
def test_gemini_thinking(self):
"""Gemini message with thoughts should return Thinking."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
msg = {
"type": "gemini",
"thoughts": "some thinking",
"content": [],
}
result = cls._extract_gemini_action(msg)
assert "Thinking" in result
def test_gemini_text_content(self):
"""Gemini message with text content should return text snippet."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
msg = {
"type": "gemini",
"content": [{"text": "Hello from Gemini"}],
}
result = cls._extract_gemini_action(msg)
assert "Hello from Gemini" in result
# =============================================
# EXTRACT TEXT FROM CONTENT TESTS
# =============================================
class TestExtractTextFromContent:
"""Tests for _extract_text_from_content static method."""
def test_string_content(self):
"""String content should return the text."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_text_from_content("Hello world")
assert "Hello world" in result
def test_list_content_with_text(self):
"""List content with text dict should return the text."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_text_from_content([{"text": "some text"}])
assert "some text" in result
def test_empty_string(self):
"""Empty string should return default."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_text_from_content("")
assert "Agent response" in result
def test_empty_list(self):
"""Empty list should return default."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_text_from_content([])
assert "Agent response" in result
def test_list_with_empty_text(self):
"""List with empty text dict should return default."""
mod, _, _, _, _ = _import_filesystem_handler()
cls = mod.MonitoringFileHandler
result = cls._extract_text_from_content([{"text": " "}])
assert "Agent response" in result
# =============================================
# PARSE GEMINI ACTIVITY TESTS
# =============================================
class TestParseGeminiActivity:
"""Tests for _parse_gemini_activity."""
def test_parses_new_messages(self):
"""Should parse new messages and emit agent event."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
session_data = json.dumps(
{
"messages": [
{"type": "user"},
{
"type": "gemini",
"content": [{"text": "Hello"}],
},
]
}
)
with patch("builtins.open", _mopen(read_data=session_data)):
result = handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert result is True
queue.enqueue.assert_called()
def test_no_new_messages(self):
"""Should return True without emitting when no new messages."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
handler._jsonl_positions["/fake/session.json"] = 2
session_data = json.dumps(
{
"messages": [
{"type": "user"},
{"type": "gemini", "content": "hi"},
]
}
)
with patch("builtins.open", _mopen(read_data=session_data)):
result = handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert result is True
queue.enqueue.assert_not_called()
def test_deduplicates_action(self):
"""Should not emit duplicate actions."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
handler._last_agent_action["/fake/session.json"] = "User message"
session_data = json.dumps({"messages": [{"type": "user"}]})
with patch("builtins.open", _mopen(read_data=session_data)):
result = handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert result is True
def test_extracts_model(self):
"""Should extract model from gemini messages."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
session_data = json.dumps(
{
"messages": [
{
"type": "gemini",
"model": "gemini-3-flash-preview",
"content": "hi",
}
]
}
)
with patch("builtins.open", _mopen(read_data=session_data)):
handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert "gemini" in handler._session_models.get("/fake/session.json", "")
def test_json_error_returns_false(self):
"""Should return False on JSON decode error."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
mock_path = MagicMock()
mock_path.name = "session.json"
with patch("builtins.open", _mopen(read_data="not json at all")):
result = handler._parse_gemini_activity(mock_path, "GEMINI")
assert result is False
def test_os_error_returns_false(self):
"""Should return False on OSError."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
mock_path = MagicMock()
mock_path.name = "session.json"
with patch("builtins.open", side_effect=OSError("fail")):
result = handler._parse_gemini_activity(mock_path, "GEMINI")
assert result is False
def test_no_queue_still_returns_true(self):
"""Should return True even when no queue."""
mod, _, _, _, _ = _import_filesystem_handler()
handler = _make_handler(mod, queue=None)
session_data = json.dumps(
{
"messages": [
{
"type": "gemini",
"content": [{"text": "hello"}],
}
]
}
)
with patch("builtins.open", _mopen(read_data=session_data)):
result = handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert result is True
def test_skips_none_actions(self):
"""Should skip messages that produce None actions."""
mod, _, _, _, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
session_data = json.dumps({"messages": [{"type": "system"}]})
with patch("builtins.open", _mopen(read_data=session_data)):
result = handler._parse_gemini_activity("/fake/session.json", "GEMINI")
assert result is True
queue.enqueue.assert_not_called()
# =============================================
# READ NEW JSONL LINES TESTS
# =============================================
@@ -1948,24 +1607,6 @@ class TestHandleEvent:
)
mock_parse.assert_called_once()
def test_gemini_json(self):
"""Should parse Gemini JSON session files."""
mod, _, mock_bd, mock_filters, queue = _import_filesystem_handler()
handler = _make_handler(mod, queue=queue)
mock_filters.should_monitor.return_value = True
mock_bd.detect_branch_from_path.return_value = "GEMINI"
with patch.object(handler, "_get_gemini_branch", return_value="PRAX"):
with patch.object(
handler,
"_parse_gemini_activity",
return_value=True,
) as mock_parse:
handler._handle_event(
"modified",
"/home/user/.gemini/tmp/slug/chats/session.json",
)
mock_parse.assert_called_once()
def test_exception_caught(self):
"""Should catch exceptions and log error."""
mod, _, _, mock_filters, queue = _import_filesystem_handler()
src/aipass/prax/tests/test_monitor_module.py
-13
View File
@@ -1409,16 +1409,3 @@ class TestGetWatchDirectoriesEdgeCases:
paths = [p for p, _r in result]
assert codex_sessions in paths
def test_includes_gemini_tmp(self, tmp_path):
"""~/.gemini/tmp is included when it exists."""
mod = _import_monitor()
fakehome = tmp_path / "fakehome"
gemini_tmp = fakehome / ".gemini" / "tmp"
gemini_tmp.mkdir(parents=True)
with patch("pathlib.Path.home", return_value=fakehome):
result = mod._get_watch_directories(tmp_path)
paths = [p for p, _r in result]
assert gemini_tmp in paths