Merge pull request #580 from AIOSAI/fix-ruff-f841-unused-variable-in-testdev

fix: ruff F841 unused variable in test_devpulse.py
This commit is contained in:
AIPass
2026-05-15 13:24:28 -07:00
committed by GitHub
43 changed files with 1240 additions and 1940 deletions
+8 -50
View File
@@ -55,9 +55,7 @@ src/aipass/{name}/
└── README.md
```
Secrets at `~/.secrets/aipass/` — API keys, tokens, credentials.
11 core branches: drone, seedgo, prax, cli, flow, ai_mail, api, trigger, spawn, memory, devpulse.
12 core branches: aipass, drone, seedgo, prax, cli, flow, ai_mail, api, trigger, spawn, memory, devpulse.
# Commands
@@ -79,11 +77,11 @@ Read-only awareness (all branches):
All write operations (commit, push, merge, checkout) restricted to devpulse via tier-based access. Dispatched agents build code, run tests — devpulse reviews diff, commits.
Drone runs git via Python subprocess, bypasses settings.json deny rules by design — drone is the gate.
Drone runs git via Python subprocess, bypasses settings.json deny rules by design — drone is the gate. `git_gate.py` PreToolUse hook enforces mechanically — applies to ALL sessions including dispatched agents. bypassPermissions does not skip hooks.
Local files = source of truth. Edit file → state on disk IS reality.
`git_gate.py` PreToolUse hook enforces mechanically — applies to ALL sessions including dispatched agents. bypassPermissions does not skip hooks.
Linting and formatting run automatically on commit via drone's commit handler (ruff check --fix + ruff format).
# aipass init
@@ -122,12 +120,6 @@ Read and reply:
Always reply to dispatch emails. Complete task → email back results. No silent completions.
# Feedback — Cross-Project Communication
Send feedback to devpulse from any project. Messages accumulate silently — no wake, no notification. DevPulse reads on demand. Works from any AIPass project (requires `AIPASS_HOME`).
Sender auto-detected. `drone @devpulse feedback --help` for commands.
# Plans (flow)
Plans manage context you don't need to carry. You don't remember what's in a plan — you remember it exists and where to find it. Registry = catalog.
@@ -140,12 +132,7 @@ Plans manage context you don't need to carry. You don't remember what's in a pla
- Other types may exist — `drone @flow --help` for current list.
Commands:
- `drone @flow create . "Subject"` — create FPLAN in current branch
- `drone @flow create /path/to "Subject"` — create FPLAN at any path
- `drone @flow create . "Subject" dplan` — create DPLAN
- `drone @flow create . "Subject" tdplan` — create TDPLAN
- `drone @flow create . "Subject" master` — create FPLAN master
- `drone @flow create . "Subject" aplan` — create APLAN
- `drone @flow create <path> "Subject" [type]` — create plan. Types: `dplan`, `aplan`, `tdplan`, `master`. Default = FPLAN. Path `.` = current branch.
- `drone @flow list open` — list active plans
- `drone @flow close <id>` — close a plan
- `drone @flow --help` — full flow reference
@@ -175,28 +162,7 @@ Where to put what:
Save proactively. Triggers: after milestone, decision, learning, before switching topics.
Archive:
- `drone @memory search <query>` — search archived memories
- `drone @memory --help` — full memory reference
# Git Workflow
Drone = only git interface. All git/gh commands denied at project level.
Read-only via drone:
- `drone @git status` — changes in your branch directory
- `drone @git diff` — actual diff
- `drone @git log` — recent commits
Write operations restricted to devpulse. You build + test → devpulse reviews + commits.
Before submitting code:
```
ruff check --fix src/ tests/
ruff format src/ tests/
```
Respect .gitignore. Gitignored patterns (`.trinity/`, `.ai_mail.local/`, `DPLAN-*`, `*.local.*`, `logs/`, `.chroma/`) ignored for a reason.
When local.json overflows limits, memories roll over to vector store via `@memory`. Search past context with `drone @memory search <query>`. `drone @memory --help` for full reference.
# How to Work
@@ -220,12 +186,13 @@ Errors go to both. Console tells user. Log tells next session.
Logs = first diagnostic tool. Check `logs/` before anything else. Don't write debug scripts or print statements — read logs.
Each branch also has `{branch}_json/` — structured JSON files per handler (config, data, log). Contains operation history, handler configuration, and runtime data. Check these for handler-level debugging alongside prax logs.
# Hard Rules
- No cross-branch file edits. Issue in another branch → email them.
- No bare imports. Always `from aipass.{module}.apps.modules...`
- No hardcoded paths. Use `Path(__file__).parents[N]` or drone for resolution.
- Never move/archive/delete files with user's name. Personal files off-limits.
- No deleting files. Rename `my_handler(disabled).py`, move to sibling `.archive/`. `(disabled)` tag gitignored. Never truly delete.
- Verify after fixing. Run test or command to confirm. Don't say "fixed" until verified.
- Cross-platform. Public package — Linux, macOS, Windows. `pathlib.Path` not string concat. `Path.home()` not `~`.
@@ -243,14 +210,5 @@ Prompts: plant breadcrumbs, not encyclopedias. Two lines ("this exists, look her
Prompts are signposts, not journals. Injected every turn — keep minimal. Never track state/sessions/context in prompts. State → `.trinity/` + `STATUS.local.md`. Prompts guide; memories record; registries catalog.
# Setup: if drone commands fail
If `drone` can't find the AIPass registry, set `AIPASS_HOME=/path/to/AIPass` in shell profile and `~/.claude/settings.json` env block.
If `drone` cannot find AIPass registry:
`export AIPASS_HOME=/path/to/AIPass`
Add to shell profile (`~/.bashrc`/`~/.zshrc`) and `~/.claude/settings.json` env block.
# Claude Code Docs (Local)
`/docs` to list topics, `/docs <topic>` to read (e.g. `/docs hooks`).
@@ -1,108 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/.aipass/aipass_global_prompt.md -->
# AIPASS — Project Context
<!-- Injected every turn via hook. -->
## What is AIPass
AIPass is a multi-agent framework. Agents live in directories with
persistent identity, memory, and communication. All AIPass infrastructure
is available from any project via the `drone` command.
## Terminology
- **Project** — this directory. Contains a registry and agents.
- **Agent** — a citizen with identity (`.trinity/`), memory, mailbox,
and code (`apps/`).
- **Registry** — `AIPASS_REGISTRY.json` tracks all agents.
## Setup: if drone commands fail
If `drone` cannot find the AIPass registry, set the env var:
```bash
export AIPASS_HOME=/path/to/AIPass # path to AIPass installation
```
Add to your shell profile (`~/.bashrc` or `~/.zshrc`) to make it permanent.
## Commands
### Agent Lifecycle
```
aipass init agent <name> # Create a new agent in src/<name>/
drone @spawn create <name> # Create agent (alternative)
drone @spawn list # List registered agents
```
### Standards
```
drone @seedgo audit <project> # Run full standards audit
drone @seedgo checklist <file> # Check a single file
```
### Dispatch — Send Task + Wake an Agent (DEFAULT)
```
drone @ai_mail dispatch @<agent> "Subject" "Body" # Send + wake (default)
drone @ai_mail dispatch @<agent> "Subject" "Body" --fresh # Send + wake fresh session
drone @ai_mail dispatch wake @<agent> # Wake without sending
drone @ai_mail dispatch wake --fresh @<agent> # Wake fresh
drone @ai_mail email @<agent> "Subject" "Body" # FYI only (no wake)
```
Use `dispatch` by default. Use `email` only when you don't need the agent to act now.
### Communication (ai_mail)
```
drone @ai_mail inbox # Check your mailbox
drone @ai_mail view <id> # Read a message
drone @ai_mail close <id> # Mark message read
```
### Feedback
```
drone @devpulse feedback send "Subject" "Body" # Send feedback (cross-project)
```
### Plans (flow)
```
drone @flow create . "Subject" dplan # Create DPLAN (design/thinking)
drone @flow create . "Subject" master # Create FPLAN master (execution)
drone @flow create . "Subject" aplan # Create APLAN (agent-level task)
drone @flow list open # List active plans
drone @flow list # List all plans
drone @flow close <id> # Close a plan
drone @flow info <id> # View plan details
```
**DPLAN** = Dev Plan. Thinking, brainstorming, architecture decisions. Use before building.
**FPLAN** = Flow Plan. Building and executing. Use when the plan is clear and work is underway.
### Memory
```
drone @memory archive # Archive memories to vector store
drone @memory search <query> # Search archived memories
```
### Git
```
drone @git status # Git status (branch-scoped)
drone @git diff # See changes in your branch
drone @git log # Recent commits
```
### Infrastructure
```
drone systems # List all available infrastructure
drone --help # Full drone command reference
```
## Patterns
- **Communication** — agents communicate via `.ai_mail.local/`.
- **Standards** — run `drone @seedgo audit` to check compliance.
- **Identity** — agents have `.trinity/passport.json`. Projects use the registry.
- **Memory** — update `.trinity/local.json` at session end. Memory is presence.
## Maintenance
- **Upgrade scaffold**: `drone @cli aipass init update` refreshes managed project files (hooks, prompts, settings) to latest templates.
- **Entry point**: each agent's `apps/{name}.py` auto-configures `sys.path` and `AIPASS_BRANCH_NAME` env var. If prax logs to `unknown_branch/`, check that these are set.
- **Standalone projects** use `src/{name}/` layout (not `src/aipass/{name}/`). Module discovery adapts automatically.
@@ -1,109 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/status/.aipass/aipass_global_prompt.md -->
# STATUS — Project Context
<!-- Injected every turn via hook. -->
## What is AIPass
AIPass is a multi-agent framework. Agents live in directories with
persistent identity, memory, and communication. All AIPass infrastructure
is available from any project via the `drone` command.
## Terminology
- **Project** — this directory. Contains a registry and agents.
- **Agent** — a citizen with identity (`.trinity/`), memory, mailbox,
and code (`apps/`).
- **Registry** — `STATUS_REGISTRY.json` tracks all agents.
## Setup: if drone commands fail
If `drone` cannot find the AIPass registry, set the env var:
```bash
export AIPASS_HOME=/path/to/AIPass # path to AIPass installation
```
Add to your shell profile (`~/.bashrc` or `~/.zshrc`) to make it permanent.
## Commands
### Agent Lifecycle
```
aipass init agent <name> # Create a new agent in src/<name>/
drone @spawn create <name> # Create agent (alternative)
drone @spawn list # List registered agents
```
### Standards
```
drone @seedgo audit <project> # Run full standards audit
drone @seedgo checklist <file> # Check a single file
```
### Dispatch — Send Task + Wake an Agent (DEFAULT)
```
drone @ai_mail dispatch @<agent> "Subject" "Body" # Send + wake (default)
drone @ai_mail dispatch @<agent> "Subject" "Body" --fresh # Send + wake fresh session
drone @ai_mail dispatch wake @<agent> # Wake without sending
drone @ai_mail dispatch wake --fresh @<agent> # Wake fresh
drone @ai_mail email @<agent> "Subject" "Body" # FYI only (no wake)
```
Use `dispatch` by default. Use `email` only when you don't need the agent to act now.
### Communication (ai_mail)
```
drone @ai_mail inbox # Check your mailbox
drone @ai_mail view <id> # Read a message
drone @ai_mail close <id> # Mark message read
```
### Feedback
```
drone @devpulse feedback send "Subject" "Body" # Send feedback (cross-project)
```
### Plans (flow)
```
drone @flow create . "Subject" dplan # Create DPLAN (design/thinking)
drone @flow create . "Subject" master # Create FPLAN master (execution)
drone @flow create . "Subject" aplan # Create APLAN (agent-level task)
drone @flow list open # List active plans
drone @flow list # List all plans
drone @flow close <id> # Close a plan
drone @flow info <id> # View plan details
```
**DPLAN** = Dev Plan. Thinking, brainstorming, architecture decisions. Use before building.
**FPLAN** = Flow Plan. Building and executing. Use when the plan is clear and work is underway.
### Memory
```
drone @memory archive # Archive memories to vector store
drone @memory search <query> # Search archived memories
```
### Git Workflow
```
drone @git status # Git status (branch-scoped)
drone @git diff # See changes
drone @git log # Commit history
```
You have no git write access. Devpulse handles all commits and PRs.
### Infrastructure
```
drone systems # List all available infrastructure
drone --help # Full drone command reference
```
## Patterns
- **Communication** — agents communicate via `.ai_mail.local/`.
- **Standards** — run `drone @seedgo audit` to check compliance.
- **Identity** — agents have `.trinity/passport.json`. Projects use the registry.
- **Memory** — update `.trinity/local.json` at session end. Memory is presence.
## Maintenance
- **Upgrade scaffold**: `drone @cli aipass init update` refreshes managed project files (hooks, prompts, settings) to latest templates.
- **Entry point**: each agent's `apps/{name}.py` auto-configures `sys.path` and `AIPASS_BRANCH_NAME` env var. If prax logs to `unknown_branch/`, check that these are set.
- **Standalone projects** use `src/{name}/` layout (not `src/aipass/{name}/`). Module discovery adapts automatically.
@@ -1,56 +0,0 @@
# Project-Level Hooks
These hooks are provisioned by `aipass init` and live in the project's
`.claude/settings.json`. They fire when CWD is inside this project.
## What fires and what doesn't
**UserPromptSubmit** hooks fire from project settings. These work:
- `branch_prompt_loader.py` — injects branch-specific prompt
- `email_notification.py` — shows unread email count
- `identity_injector.py` — injects branch identity from passport
**PreToolUse / PostToolUse** hooks are provisioned but **DO NOT FIRE** from
project-level settings. This is a Claude Code limitation (confirmed S122,
GitHub issue #36071). These scripts exist but are dead weight:
- `pre_edit_gate.py` — intended to block cross-branch writes (never runs)
- `auto_fix_diagnostics.py` — intended to run pyright+ruff (never runs)
- `subagent_stop_gate.py` — intended to check subagent files (never runs)
These same hooks DO fire from provider settings (`~/.claude/settings.json`)
where they are also wired. The provider copies handle all enforcement.
**PreCompact** hooks fire from project settings:
- `pre_compact.py` — injects recovery context after compaction
## CWD guard interaction
When this project has UserPromptSubmit hooks (it does), the provider-level
UserPromptSubmit hooks detect this and exit silently. This prevents the AIPass
global prompt from being injected into projects that manage their own context.
The provider-level PreToolUse/PostToolUse hooks still fire (they can only run
at provider level) — so enforcement (git_gate, pre_edit_gate, auto_fix) is
always active regardless of CWD.
## Testing
Provider-level test harness covers project-level behavior:
```bash
python3 $AIPASS_HOME/.claude/hooks/hook_test.py --direct
```
Tests include:
- `direct_provider_guards_for_init_project` — verifies provider hooks are
CWD-guarded when run from an aipass init project
- `direct_project_settings_schema` — validates project settings.json has
expected hooks and all referenced scripts exist
## Updating hooks
```bash
drone @cli aipass init update # Refresh managed project files to latest templates
```
## Related
See `$AIPASS_HOME/.claude/hooks/README.md` for the full hook system documentation.
@@ -1,366 +0,0 @@
#!/usr/bin/env python3
"""
PostToolUse Auto-fix Hook — Detects errors and surfaces them for fixing.
Two-hook system:
PostToolUse (this file) → runs pyright + ruff on edited file, saves errors to state
PreToolUse (pre_edit_gate.py) → blocks edits to OTHER files until errors fixed
Key behaviors:
- Runs py_compile (syntax), ruff lint+format, pyright (type errors) on edited file
- Runs seedgo checklist for AIPass standards
- Saves ruff lint AND pyright errors to state file for PreToolUse gate (hard block)
- Surfaces ALL errors in additionalContext so Claude sees them
Version: 5.2.0
CHANGELOG:
- v5.2.0 (2026-04-20): Save ruff lint errors to state file for hard-block enforcement.
Pre-edit gate now blocks on F401/lint just like type errors.
- v5.1.0 (2026-04-19): Added ruff format --check to surface format drift.
- v5.0.0 (2026-03-17): Replaced mcp__ide__getDiagnostics with direct pyright.
Added state file for PreToolUse gate integration.
Single-file pyright (not whole project).
- v4.3.0 (2026-03-17): Added seedgo checklist integration
- v4.0.0 (2025-11-27): Complete rewrite - actual validation, silent operation
"""
import json
import sys
import subprocess
from pathlib import Path
EDIT_TOOLS = ["Edit", "Write", "MultiEdit", "NotebookEdit"]
LAST_FILE_PATH = Path(__file__).parent / ".last_diagnostics_file"
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
SKIP_EXTENSIONS = {".md", ".txt", ".log", ".csv", ".html"}
# AIPass-specific Python patterns to check
PYTHON_PATTERNS = {
"bad_optional": {"pattern": ": str = None", "message": "Optional param should use 'str | None = None' pattern"},
"logger_debug": {
"pattern": "logger.debug(",
"message": "Use logger.info for SystemLogger (logger.debug not supported)",
},
"return_error_msg": {
"pattern": "return error_msg",
"message": "Return None for error states, not error_msg string",
},
"open_no_encoding": {
"pattern": "open(",
"requires_missing": "encoding=",
"message": "open() without encoding='utf-8'",
},
"log_not_log_operation": {
"pattern": ".log(",
"message": "Use log_operation() with success/error params, not .log()",
},
"dict_none_no_check": {
"pattern": "Dict | None",
"message": "Dict | None return: Add None check before using (if result is None: return)",
},
}
# JSON-specific patterns for emoji corruption
JSON_CORRUPTION_CHARS = ["\ufffd", "\x00"]
def run_python_checks(file_path: str) -> list[str]:
"""Run actual Python validation - returns list of errors."""
errors = []
# 1. Syntax check with py_compile
try:
result = subprocess.run(
[sys.executable, "-m", "py_compile", file_path], capture_output=True, text=True, timeout=5
)
if result.returncode != 0:
errors.append(f"SYNTAX: {result.stderr.strip()}")
except Exception:
pass
# 2. Ruff check (if available) - fast linter
try:
result = subprocess.run(
["ruff", "check", "--select=E,F,W", "--output-format=text", file_path],
capture_output=True,
text=True,
timeout=10,
)
if result.stdout.strip():
for line in result.stdout.strip().split("\n")[:5]:
errors.append(f"LINT: {line}")
except FileNotFoundError:
pass
except Exception:
pass
# 3. Ruff format check — detect format drift
try:
result = subprocess.run(["ruff", "format", "--check", file_path], capture_output=True, text=True, timeout=10)
if result.returncode != 0:
errors.append(f"FORMAT: {Path(file_path).name} needs ruff format (run: ruff format {Path(file_path).name})")
except FileNotFoundError:
pass
except Exception:
pass
# 4. AIPass-specific pattern checks
try:
content = Path(file_path).read_text(encoding="utf-8")
lines = content.split("\n")
for check in PYTHON_PATTERNS.values():
pattern = check["pattern"]
message = check["message"]
requires_missing = check.get("requires_missing")
if requires_missing:
if pattern in content and requires_missing not in content:
errors.append(f"PATTERN: {message}")
continue
for line in lines:
stripped = line.strip()
if stripped.startswith(("#", '"', "'")):
continue
if f'"{pattern}' in line or f"'{pattern}" in line:
continue
if pattern in line:
errors.append(f"PATTERN: {message}")
break
except Exception:
pass
return errors
def run_ruff_lint_structured(file_path: str) -> list[dict]:
"""Run ruff check and return structured violations for the state file.
Returns list of {line, message} dicts — same format as pyright errors.
Only non-empty when ruff finds real violations (not format drift).
"""
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
["ruff", "check", "--select=E,F,W", "--output-format=json", file_path],
capture_output=True,
text=True,
timeout=10,
)
if not result.stdout.strip():
return []
violations = json.loads(result.stdout)
if not isinstance(violations, list):
return []
errors = []
for v in violations[:10]:
line = v.get("location", {}).get("row", 0)
code = v.get("code", "?")
message = v.get("message", "unknown")[:100]
errors.append({"line": line, "message": f"{code}: {message}"})
return errors
except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired, Exception):
return []
def run_pyright_check(file_path: str) -> list[dict]:
"""Run pyright on a single file. Returns list of error dicts."""
# Skip hook files - they don't follow project standards
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
[sys.executable, "-m", "pyright", "--outputjson", file_path], capture_output=True, text=True, timeout=15
)
try:
data = json.loads(result.stdout)
except (json.JSONDecodeError, ValueError):
return []
errors = []
for diag in data.get("generalDiagnostics", []):
severity = diag.get("severity", "")
if severity == "error":
line = diag.get("range", {}).get("start", {}).get("line", 0)
message = diag.get("message", "Unknown error")
errors.append({"line": line, "message": message[:100]})
return errors[:10] # Max 10 errors
except FileNotFoundError:
return [] # pyright not installed
except subprocess.TimeoutExpired:
return [] # Timeout — don't block
except Exception:
return []
def save_diagnostics_state(file_path: str, errors: list[dict]):
"""Save type errors to state file for PreToolUse gate."""
try:
if errors:
state = {"file": str(Path(file_path).resolve()), "errors": errors}
STATE_FILE.write_text(json.dumps(state), encoding="utf-8")
else:
# No errors — clear the state
if STATE_FILE.exists():
STATE_FILE.unlink()
except Exception:
pass
def run_json_checks(file_path: str) -> list[str]:
"""Run actual JSON validation - returns list of errors."""
errors = []
try:
content = Path(file_path).read_text(encoding="utf-8")
for char in JSON_CORRUPTION_CHARS:
if char in content:
errors.append(f"EMOJI CORRUPTION: Found corrupted character '{repr(char)}'")
break
try:
data = json.loads(content)
if isinstance(data, dict):
for key in ["allowed_emojis", "emojis", "emoji_list"]:
if key in data and isinstance(data[key], list):
for item in data[key]:
if isinstance(item, str) and len(item) == 1:
if ord(item) < 128 and item not in "\u2713\u2717":
errors.append(f"EMOJI CORRUPTION: Suspicious char '{item}' in {key}")
break
except json.JSONDecodeError as e:
errors.append(f"JSON SYNTAX: {e.msg} at line {e.lineno}")
except Exception as e:
errors.append(f"READ ERROR: {e!s}")
return errors
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo standards checklist — returns violations only."""
if "/.claude/hooks/" in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True,
text=True,
timeout=15,
cwd=str(Path.home() / "Projects" / "AIPass"),
)
if result.returncode != 0:
return []
violations = []
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("\u2717"):
violation = line[1:].strip()
if violation:
violations.append(violation)
return violations[:5]
except FileNotFoundError:
return []
except Exception:
return []
def should_skip_file(file_path: str) -> bool:
"""Check if file should be skipped."""
if not file_path:
return True
ext = Path(file_path).suffix.lower()
return ext in SKIP_EXTENSIONS
def is_same_file_as_last(file_path: str) -> bool:
"""Smart batching DISABLED — always recheck.
Previously skipped rechecks on the same file, but this caused
errors introduced on second edit to be missed (state file didn't
exist from first clean edit, so skip triggered). The 1.7s pyright
cost per edit is acceptable for correctness.
"""
return False
def main():
"""Main hook entry point."""
try:
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
if tool_name not in EDIT_TOOLS:
return
if should_skip_file(file_path):
return
if is_same_file_as_last(file_path):
return
# Collect all errors
errors = []
if file_path.endswith(".py"):
errors = run_python_checks(file_path)
# Seedgo standards checklist
seedgo_violations = run_seedgo_checklist(file_path)
for v in seedgo_violations:
errors.append(f"SEEDGO: {v}")
# Pyright type errors (single file)
type_errors = run_pyright_check(file_path)
for te in type_errors:
errors.append(f"TYPE: L{te['line']}: {te['message']}")
# Save ruff lint + type errors to state file for PreToolUse gate (hard block)
ruff_lint_errors = run_ruff_lint_structured(file_path)
save_diagnostics_state(file_path, ruff_lint_errors + type_errors)
elif file_path.endswith(".json"):
errors = run_json_checks(file_path)
else:
return
# Build output
if errors:
error_text = "\n".join(f" - {e}" for e in errors)
context = f"""[AUTO-FIX] {len(errors)} error(s) in {Path(file_path).name}:
{error_text}
Fix these errors in {Path(file_path).name} now. Do not skip or defer."""
output = {
"hookSpecificOutput": {"hookEventName": "PostToolUse", "additionalContext": context},
"systemMessage": f"[AUTO-FIX] {len(errors)} error(s) — fix before continuing",
}
print(json.dumps(output))
else:
output = {"systemMessage": "[diagnostics] ok"}
print(json.dumps(output))
except Exception:
pass # Silent fail
if __name__ == "__main__":
main()
@@ -1,53 +0,0 @@
#!/usr/bin/env python3
"""
Branch Prompt Loader — AIPass Public Repo
Injects branch-specific prompts based on CWD. When working in a branch
directory, loads .aipass/aipass_local_prompt.md and outputs it so the
AI sees branch-specific context.
Version: 1.0.0
"""
from pathlib import Path
def find_branch_root() -> Path | None:
"""
Find the branch root directory.
Looks for .trinity/ or .aipass/ as branch indicators.
Stops at the repo root (has pyproject.toml or .git).
"""
cwd = Path.cwd()
search_path = cwd
while search_path.parent != search_path:
# Branch indicators: has .trinity/ (memory files) or apps/ (code)
has_trinity = (search_path / ".trinity").is_dir()
has_apps = (search_path / "apps").is_dir()
if has_trinity or has_apps:
return search_path
# Stop at repo root
if (search_path / "pyproject.toml").exists() or (search_path / ".git").is_dir():
return None
search_path = search_path.parent
return None
def main():
branch_root = find_branch_root()
if branch_root:
prompt_file = branch_root / ".aipass" / "aipass_local_prompt.md"
if prompt_file.exists():
content = prompt_file.read_text().strip()
branch_name = branch_root.name.upper()
print(f"\n# Branch Context: {branch_name}\n<!-- Source: {prompt_file} -->\n{content}")
if __name__ == "__main__":
main()
@@ -1,96 +0,0 @@
#!/usr/bin/env python3
"""
Email Notification Hook - Notifies of new emails on prompt submit.
Checks the current branch's inbox for unread emails and displays
a notification if any exist.
Version: 1.0.0
"""
import json
from pathlib import Path
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
while search.parent != search:
if (search / "pyproject.toml").exists() or (search / ".git").is_dir():
return search
search = search.parent
return None
def find_branch_root() -> Path | None:
"""Find the branch root directory by walking up from CWD."""
cwd = Path.cwd()
repo_root = find_repo_root()
if not repo_root:
return None
search_path = cwd
for _ in range(10):
has_trinity = (search_path / ".trinity").is_dir()
has_id = list(search_path.glob("*.id.json"))
has_apps = (search_path / "apps").is_dir()
has_mail = (search_path / ".ai_mail.local").is_dir() or (search_path / "ai_mail.local").is_dir()
if (has_trinity or has_id or has_apps or has_mail) and search_path != repo_root:
return search_path
if search_path == repo_root:
break
parent = search_path.parent
if parent == search_path:
break
search_path = parent
return None
def count_new_emails(branch_root: Path) -> int:
"""Count new (unread) emails in the branch's inbox."""
# Check both patterns: .ai_mail.local (canonical) and ai_mail.local (legacy)
inbox_path = branch_root / ".ai_mail.local" / "inbox.json"
if not inbox_path.exists():
inbox_path = branch_root / "ai_mail.local" / "inbox.json"
if not inbox_path.exists():
return 0
try:
with open(inbox_path, "r", encoding="utf-8") as f:
data = json.load(f)
# Handle both formats: {"messages": [...]} and bare [...]
messages = data if isinstance(data, list) else data.get("messages", [])
count = 0
for msg in messages:
if msg.get("status") == "new":
count += 1
elif msg.get("status") is None and not msg.get("read", False):
count += 1
return count
except (json.JSONDecodeError, OSError):
return 0
def main():
branch_root = find_branch_root()
if not branch_root:
return
new_count = count_new_emails(branch_root)
if new_count > 0:
plural = "s" if new_count != 1 else ""
print(
f"You have {new_count} new email{plural} - check with: drone @ai_mail inbox | then: drone @ai_mail view <id> | close with: drone @ai_mail close <id>"
)
if __name__ == "__main__":
main()
@@ -1,118 +0,0 @@
#!/usr/bin/env python3
"""
Identity Injector - Injects branch identity on every prompt.
Reads from [BRANCH].id.json and outputs core identity fields.
Finds the branch root by walking up from CWD looking for apps/ or *.id.json.
Version: 1.0.0
"""
import json
from pathlib import Path
def find_repo_root() -> Path | None:
"""Find the repo root (contains pyproject.toml or .git)."""
search = Path.cwd()
while search.parent != search:
if (search / "pyproject.toml").exists() or (search / ".git").is_dir():
return search
search = search.parent
return None
def find_branch_root() -> Path | None:
"""Find the branch root directory by walking up from CWD."""
cwd = Path.cwd()
repo_root = find_repo_root()
if not repo_root:
return None
search_path = cwd
while search_path >= repo_root:
has_trinity = (search_path / ".trinity").is_dir()
has_id = list(search_path.glob("*.id.json"))
if has_trinity or has_id:
return search_path
if search_path == repo_root:
break
search_path = search_path.parent
return None
def find_id_file(branch_root: Path) -> Path | None:
"""Find the identity file for a branch (.trinity/passport.json or *.id.json)."""
# AIPass pattern: .trinity/passport.json
passport = branch_root / ".trinity" / "passport.json"
if passport.exists():
return passport
# Dev-Pass fallback: *.id.json
id_files = list(branch_root.glob("*.id.json"))
if id_files:
return id_files[0]
return None
def format_identity(data: dict) -> str:
"""Format branch_info + identity for injection."""
lines = []
# Try branch_info first (enriched passports), fall back to identity block (setup.sh passports)
branch = data.get("branch_info", {})
identity = data.get("identity", {})
name = branch.get("branch_name") or identity.get("name", "UNKNOWN")
lines.append(f"# {name} Identity")
lines.append(f"Path: {branch.get('path', 'unknown')}")
lines.append(f"Email: {branch.get('email', 'unknown')}")
identity = data.get("identity", {})
if identity.get("role"):
lines.append(f"Role: {identity['role']}")
traits = identity.get("traits") or data.get("traits")
if traits:
if isinstance(traits, list):
lines.append("Traits: " + " | ".join(traits))
else:
lines.append(f"Traits: {traits}")
if identity.get("purpose"):
lines.append(f"Purpose: {identity['purpose']}")
what_i_do = identity.get("what_i_do", [])
if what_i_do:
lines.append("Do: " + " | ".join(what_i_do[:4]))
what_i_dont_do = identity.get("what_i_dont_do", [])
if what_i_dont_do:
lines.append("Don't: " + " | ".join(what_i_dont_do[:3]))
principles = data.get("principles", [])
if principles:
lines.append("Principles: " + " * ".join(principles))
return "\n".join(lines)
def main():
branch_root = find_branch_root()
if not branch_root:
return
id_file = find_id_file(branch_root)
if not id_file or not id_file.exists():
return
try:
data = json.loads(id_file.read_text(encoding="utf-8"))
output = format_identity(data)
if output:
print(f"\n{output}")
except (json.JSONDecodeError, KeyError):
pass
if __name__ == "__main__":
main()
@@ -1,168 +0,0 @@
#!/usr/bin/env python3
"""
Pre-Compact Hook - Inject live state for post-compact recovery.
Reads STATUS.local.md, last session from local.json, and git branch
to give the model real context after compaction — not generic advice.
Version: 3.0.0
"""
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,
)
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():
"""Main hook entry point."""
try:
json.load(sys.stdin)
branch_dir = _find_branch_dir()
branch_name = _get_branch_name(branch_dir)
sections = []
sections.append(f"""POST-COMPACT RECOVERY — @{branch_name}
Context just compacted. Below is your live state. Use it to continue seamlessly.""")
# Git info
git_info = _get_git_info()
if git_info:
sections.append(f"## Git\n{git_info}")
# 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}")
# 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
- .trinity/local.json has full session history and key_learnings — read it if you need more context
- STATUS.local.md has current work, known issues, and todos
- 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)
if __name__ == "__main__":
main()
@@ -1,149 +0,0 @@
#!/usr/bin/env python3
"""
PreToolUse Gate — Blocks unsafe edits at the hook layer.
Rules (checked in order):
1. Inbox lock — any write targeting *.ai_mail.local/inbox.json is BLOCKED.
Use `drone @ai_mail email` instead.
2. Cross-branch — writes to src/aipass/X/** from a CWD inside src/aipass/Y/**
are BLOCKED unless the calling branch is in TRUSTED_CROSS_WRITERS.
3. State-file — edits to OTHER .py files while the current branch has unresolved
type errors are BLOCKED. (original v1.2.0 logic)
Track E additions: rules 1 + 2 (DPLAN-0139).
Version: 1.3.0
"""
import json
import os
import sys
from pathlib import Path
STATE_FILE = Path(__file__).parent / ".diagnostics_state.json"
EDIT_TOOLS = {"Edit", "Write", "MultiEdit", "NotebookEdit"}
# Single source of truth lives in permissions.py — inline here as fallback
# so the hook works even when aipass package is not on sys.path.
TRUSTED_CROSS_WRITERS: tuple[str, ...] = ("devpulse", "seedgo", "spawn")
def _get_branch(file_path: str) -> str:
"""Extract AIPass branch name from a file path (src/aipass/{branch}/ pattern)."""
parts = Path(file_path).parts
for i, part in enumerate(parts):
if part == "aipass" and i > 0 and parts[i - 1] == "src" and i + 1 < len(parts):
return parts[i + 1]
return ""
def _block(reason: str) -> None:
# codeql[py/clear-text-logging-sensitive-data]
print(json.dumps({"decision": "block", "reason": reason}))
sys.exit(2)
def main():
try:
input_data = json.load(sys.stdin)
tool_name = input_data.get("tool_name", "")
tool_input = input_data.get("tool_input", {})
file_path = tool_input.get("file_path", "")
if tool_name not in EDIT_TOOLS:
return
if not file_path:
return
# ------------------------------------------------------------------
# Rule 1: Inbox lock — block all writes to *.ai_mail.local/inbox.json
# ------------------------------------------------------------------
fp = Path(file_path)
if fp.name == "inbox.json" and ".ai_mail.local" in fp.parts:
_block('Direct writes to inbox.json are blocked.\nUse: drone @ai_mail email @<branch> "Subject" "Body"')
# ------------------------------------------------------------------
# Rule 1.5: Dispatched-agent path confinement (DPLAN-0155 M3)
# Daemon-spawned agents can only write inside their own branch dir.
# Breaks the prompt-injection amplifier chain — even if injected,
# a dispatched agent cannot write to other agents' inboxes or code.
# ------------------------------------------------------------------
cwd = input_data.get("cwd", "") or os.getcwd()
cwd_branch = _get_branch(cwd)
session_type = os.environ.get("AIPASS_SESSION_TYPE", "interactive")
if session_type == "daemon" and cwd_branch:
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if target_branch and target_branch != cwd_branch:
_block(
f"Dispatched agent confined to own branch: '{cwd_branch}' "
f"cannot write to '{target_branch}' in daemon mode."
)
repo_root = None
for parent in Path(cwd).parents:
if (parent / ".git").exists():
repo_root = parent
break
if repo_root and not target_branch:
allowed_prefix = str(repo_root / "src" / "aipass" / cwd_branch)
resolved = str(fp.resolve()) if not fp.is_absolute() else str(fp)
if not resolved.startswith(allowed_prefix):
_block(f"Dispatched agent restricted to {allowed_prefix}. Cannot write to: {file_path}")
# ------------------------------------------------------------------
# Rule 2: Cross-branch write enforcement
# ------------------------------------------------------------------
target_branch = _get_branch(str(fp.resolve()) if not fp.is_absolute() else str(fp))
if cwd_branch and target_branch and cwd_branch != target_branch:
if cwd_branch not in TRUSTED_CROSS_WRITERS:
_block(
f"Cross-branch write blocked: '{cwd_branch}' cannot write to '{target_branch}'.\n"
f"Trusted cross-writers: {', '.join(TRUSTED_CROSS_WRITERS)}"
)
# ------------------------------------------------------------------
# Rule 3: State-file (original v1.2.0) — .py files only
# ------------------------------------------------------------------
if not file_path.endswith(".py"):
return
if not STATE_FILE.exists():
return
try:
state = json.loads(STATE_FILE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, IOError):
return
errored_file = state.get("file", "")
errors = state.get("errors", [])
if not errors:
return
try:
current = str(Path(file_path).resolve())
errored = str(Path(errored_file).resolve())
except (OSError, ValueError):
return
if current == errored:
return
current_branch = _get_branch(current)
errored_branch = _get_branch(errored)
if not errored_branch:
return
if current_branch and errored_branch and current_branch != errored_branch:
return
error_summary = "\n".join(f" L{e['line']}: {e['message']}" for e in errors[:5])
_block(f"Fix {len(errors)} error(s) in {Path(errored_file).name} before editing other files:\n{error_summary}")
except Exception:
pass # Silent fail → allow
if __name__ == "__main__":
main()
@@ -1,114 +0,0 @@
#!/usr/bin/env python3
"""
SubagentStop Gate — Checks files modified by subagents before allowing them to finish.
Runs seedgo checklist + basic validation on any .py files the subagent touched.
If violations found, blocks the stop and tells the subagent to fix them.
Version: 1.0.0
"""
import json
import os
import sys
import subprocess
from pathlib import Path
def _find_repo_root() -> Path | None:
"""Walk up from CWD or AIPASS_HOME to find the git repo root."""
for start in (os.environ.get("AIPASS_HOME", ""), os.getcwd()):
p = Path(start)
while p != p.parent:
if (p / ".git").exists():
return p
p = p.parent
return None
AIPASS_ROOT = _find_repo_root()
def get_modified_py_files() -> list[str]:
"""Get Python files modified in the working tree (unstaged + staged)."""
if AIPASS_ROOT is None:
return []
try:
result = subprocess.run(
["git", "diff", "--name-only", "HEAD"], capture_output=True, text=True, timeout=5, cwd=str(AIPASS_ROOT)
)
files = []
for line in result.stdout.strip().split("\n"):
line = line.strip()
if line.endswith(".py") and not line.startswith(".claude/"):
full = AIPASS_ROOT / line
if full.exists():
files.append(str(full))
return files
except Exception:
return []
def run_seedgo_checklist(file_path: str) -> list[str]:
"""Run seedgo checklist on a single file."""
if AIPASS_ROOT is None:
return []
if "/.claude/" in file_path:
return []
try:
result = subprocess.run(
["drone", "@seedgo", "checklist", file_path],
capture_output=True,
text=True,
timeout=15,
cwd=str(AIPASS_ROOT),
)
if result.returncode != 0:
return []
violations = []
for line in result.stdout.split("\n"):
line = line.strip()
if line.startswith("\u2717"):
v = line[1:].strip()
if v:
violations.append(v)
return violations[:5]
except Exception:
return []
def main():
try:
json.load(sys.stdin)
modified = get_modified_py_files()
if not modified:
return # Nothing to check
all_violations = {}
for f in modified:
vs = run_seedgo_checklist(f)
if vs:
name = Path(f).name
all_violations[name] = vs
if not all_violations:
return # All clear
# Build the block reason
lines = ["Standards violations found in files you modified:\n"]
for fname, vs in all_violations.items():
lines.append(f" {fname}:")
for v in vs:
lines.append(f" - {v}")
lines.append("\nFix these violations before finishing.")
output = {"decision": "block", "reason": "\n".join(lines)}
print(json.dumps(output))
except Exception:
pass # Silent fail — don't block on errors
if __name__ == "__main__":
main()
@@ -1,43 +0,0 @@
{
"hooks": {
"UserPromptSubmit": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/branch_prompt_loader.py"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/email_notification.py"
}
]
},
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/identity_injector.py"
}
]
}
],
"PreCompact": [
{
"hooks": [
{
"type": "command",
"command": "python3 .claude/hooks/pre_compact.py"
}
]
}
]
},
"env": {
"AIPASS_HOME": "/home/patrick/Projects/AIPass"
}
}
-37
View File
@@ -1,37 +0,0 @@
# AIPass local state
.trinity/
.ai_mail.local/
*.local.*
!STATUS.local.md
# Plans (local working docs)
DPLAN-*
FPLAN-*
APLAN-*
TDPLAN-*
# Logs
logs/
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
.venv/
venv/
# IDE
.vscode/
.idea/
# OS
.DS_Store
Thumbs.db
# Archives
.archive/
# Disabled files
*(disabled)*
-30
View File
@@ -1,30 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/status/AGENTS.md -->
# STATUS — Agent Instructions
This project uses AIPass, a multi-agent framework.
## Key Concepts
- **Project** — this directory. Contains a registry and one or more agents.
- **Agent** — a citizen that lives inside the project with its own identity, memory, and code.
- **Registry** — `STATUS_REGISTRY.json` tracks all agents.
## Getting Started
Create your first agent:
```
aipass init agent <name>
```
## Available Commands
```
aipass init agent <name> # Create a new agent
drone @spawn create <name> # Create agent (alternative)
drone @seedgo audit <project> # Run standards audit
drone systems # List all infrastructure
```
## Startup
On startup, read: `STATUS_REGISTRY.json`, `README.md`, `STATUS.local.md`
-27
View File
@@ -1,27 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/aipass/status/GEMINI.md -->
# STATUS — Project Instructions
This project uses AIPass, a multi-agent framework.
## Key Concepts
- **Project** — this directory. Contains a registry and one or more agents.
- **Agent** — a citizen that lives inside the project with its own identity, memory, and code.
- **Registry** — `STATUS_REGISTRY.json` tracks all agents.
## Getting Started
Create your first agent: `aipass init agent <name>`
## Available Commands
```
aipass init agent <name> # Create a new agent
drone @spawn create <name> # Create agent (alternative)
drone @seedgo audit <project> # Run standards audit
drone systems # List all infrastructure
```
## Startup
On startup, read: `STATUS_REGISTRY.json`, `README.md`, `STATUS.local.md`
-53
View File
@@ -1,53 +0,0 @@
# STATUS
An AIPass project.
## Quick Start
```bash
# 1. Create your first agent
aipass init agent my_agent
# 2. Start a session
cd src/my_agent/
claude # or your preferred AI CLI
# 3. Check project status
cat STATUS.local.md
```
## Project Structure
```
status/
STATUS_REGISTRY.json # Agent registry
.aipass/ # Prompts (injected per-turn)
CLAUDE.md # Claude Code instructions
AGENTS.md # Codex instructions
GEMINI.md # Gemini instructions
STATUS.local.md # Project status
src/ # Agent directories live here
<agent_name>/ # Created via aipass init agent
```
## What is AIPass?
AIPass is a multi-agent framework where autonomous agents (citizens) live in directories with persistent identity, memory, and communication.
Each agent has:
- **Identity** — `.trinity/passport.json`
- **Memory** — `.trinity/local.json`, `observations.json`
- **Mailbox** — `.ai_mail.local/`
- **Code** — `apps/` with modules and handlers
## Commands
| Command | Description |
|---------|-------------|
| `aipass init agent <name>` | Create a new agent |
| `drone @spawn create <name>` | Create agent (alternative) |
| `drone @seedgo audit <project>` | Run standards audit |
| `drone @ai_mail inbox` | Check agent mailbox |
| `drone systems` | List infrastructure |
*Initialized with [AIPass](https://github.com/AIOSAI/AIPass) on 2026-05-04*
-9
View File
@@ -1,9 +0,0 @@
# STATUS
**State:** New
**Last update:** 2026-05-04
## Current Work
## Known Issues
- None
@@ -1,108 +0,0 @@
<!-- Source: /home/patrick/Projects/AIPass/src/aipass/devpulse/.aipass/aipass_global_prompt.md -->
# DEVPULSE — Project Context
<!-- Injected every turn via hook. -->
## What is AIPass
AIPass is a multi-agent framework. Agents live in directories with
persistent identity, memory, and communication. All AIPass infrastructure
is available from any project via the `drone` command.
## Terminology
- **Project** — this directory. Contains a registry and agents.
- **Agent** — a citizen with identity (`.trinity/`), memory, mailbox,
and code (`apps/`).
- **Registry** — `DEVPULSE_REGISTRY.json` tracks all agents.
## Setup: if drone commands fail
If `drone` cannot find the AIPass registry, set the env var:
```bash
export AIPASS_HOME=/path/to/AIPass # path to AIPass installation
```
Add to your shell profile (`~/.bashrc` or `~/.zshrc`) to make it permanent.
## Commands
### Agent Lifecycle
```
aipass init agent <name> # Create a new agent in src/<name>/
drone @spawn create <name> # Create agent (alternative)
drone @spawn list # List registered agents
```
### Standards
```
drone @seedgo audit <project> # Run full standards audit
drone @seedgo checklist <file> # Check a single file
```
### Dispatch — Send Task + Wake an Agent (DEFAULT)
```
drone @ai_mail dispatch @<agent> "Subject" "Body" # Send + wake (default)
drone @ai_mail dispatch @<agent> "Subject" "Body" --fresh # Send + wake fresh session
drone @ai_mail dispatch wake @<agent> # Wake without sending
drone @ai_mail dispatch wake --fresh @<agent> # Wake fresh
drone @ai_mail email @<agent> "Subject" "Body" # FYI only (no wake)
```
Use `dispatch` by default. Use `email` only when you don't need the agent to act now.
### Communication (ai_mail)
```
drone @ai_mail inbox # Check your mailbox
drone @ai_mail view <id> # Read a message
drone @ai_mail close <id> # Mark message read
```
### Feedback
```
drone @devpulse feedback send "Subject" "Body" # Send feedback (cross-project)
```
### Plans (flow)
```
drone @flow create . "Subject" dplan # Create DPLAN (design/thinking)
drone @flow create . "Subject" master # Create FPLAN master (execution)
drone @flow create . "Subject" aplan # Create APLAN (agent-level task)
drone @flow list open # List active plans
drone @flow list # List all plans
drone @flow close <id> # Close a plan
drone @flow info <id> # View plan details
```
**DPLAN** = Dev Plan. Thinking, brainstorming, architecture decisions. Use before building.
**FPLAN** = Flow Plan. Building and executing. Use when the plan is clear and work is underway.
### Memory
```
drone @memory archive # Archive memories to vector store
drone @memory search <query> # Search archived memories
```
### Git
```
drone @git status # Git status (branch-scoped)
drone @git diff # See changes in your branch
drone @git log # Recent commits
```
### Infrastructure
```
drone systems # List all available infrastructure
drone --help # Full drone command reference
```
## Patterns
- **Communication** — agents communicate via `.ai_mail.local/`.
- **Standards** — run `drone @seedgo audit` to check compliance.
- **Identity** — agents have `.trinity/passport.json`. Projects use the registry.
- **Memory** — update `.trinity/local.json` at session end. Memory is presence.
## Maintenance
- **Upgrade scaffold**: `drone @cli aipass init update` refreshes managed project files (hooks, prompts, settings) to latest templates.
- **Entry point**: each agent's `apps/{name}.py` auto-configures `sys.path` and `AIPASS_BRANCH_NAME` env var. If prax logs to `unknown_branch/`, check that these are set.
- **Standalone projects** use `src/{name}/` layout (not `src/aipass/{name}/`). Module discovery adapts automatically.
+24 -171
View File
@@ -7,200 +7,53 @@
"bypass": [
{
"standard": "architecture",
"reason": "No 'manager' citizen_class template exists in spawn. Devpulse is the only manager branch — not a missing file, just an unsupported template type."
},
{
"standard": "log_structure",
"reason": "Manager branch — coordinates via dispatch, doesn't generate system logs directly. Local logs from prax are sufficient."
},
{
"standard": "test_quality",
"reason": "Manager branch with minimal apps/ (auto-generated by spawn template push). Test quality improvements deferred until devpulse has real module logic to test."
},
{
"standard": "cli_flags",
"reason": "Devpulse is not a user-facing CLI module — it's an orchestration hub accessed via drone routing, not direct invocation."
},
{
"standard": "meta",
"file": "apps/devpulse.py",
"reason": "META block present but checker format mismatch — auto-generated entry point from spawn template push."
},
{
"standard": "encapsulation",
"file": "apps/devpulse.py",
"reason": "Handler security guard present (handle_command with caller check) but checker pattern mismatch."
},
{
"standard": "stderr_routing",
"file": "apps/devpulse.py",
"reason": "Manager branch uses Rich Console directly — no cli_service dependency needed for a branch that doesn't build CLI features."
},
{
"standard": "cli",
"file": "apps/devpulse.py",
"reason": "Manager branch — not a CLI module. Entry point auto-generated by spawn template. --help is implemented via print_introspection()."
},
{
"standard": "encapsulation",
"file": "apps/modules/watchdog.py",
"reason": "Watchdog module has its own _guard_caller() that blocks cross-branch invocation at the handle_command boundary. Inherits branch-level gap: devpulse has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "json_structure",
"file": "apps/modules/watchdog.py",
"reason": "Devpulse is a manager branch with no apps/handlers/json/json_handler. Watchdog logs through prax system_logger instead."
},
{
"standard": "encapsulation",
"file": "apps/handlers/watchdog/agent.py",
"reason": "Inherits branch-level gap: devpulse has no apps/handlers/__init__.py inspect.stack guard (manager branch). Cross-branch protection enforced at the module layer via _guard_caller()."
},
{
"standard": "json_structure",
"file": "apps/handlers/watchdog/agent.py",
"reason": "Devpulse is a manager branch with no json_handler. Agent handler logs through prax system_logger to stderr."
},
{
"standard": "encapsulation",
"file": "apps/handlers/watchdog/__init__.py",
"reason": "Structural Python package marker — inherits branch-level gap (devpulse has no apps/handlers/__init__.py inspect.stack guard, manager branch)."
},
{
"standard": "naming",
"file": "apps/handlers/watchdog/__init__.py",
"reason": "Standard Python package marker filename — required by Python, cannot be renamed to snake_case."
},
{
"standard": "encapsulation",
"file": "apps/handlers/watchdog/timer.py",
"reason": "Inherits branch-level gap: devpulse has no apps/handlers/__init__.py inspect.stack guard (manager branch). Cross-branch protection enforced at the module layer via watchdog.py _guard_caller(). Same situation as watchdog/agent.py."
},
{
"standard": "json_structure",
"file": "apps/handlers/watchdog/timer.py",
"reason": "Devpulse is a manager branch with no json_handler. Timer handler logs through prax system_logger. Same situation as watchdog/agent.py."
},
{
"standard": "encapsulation",
"file": "apps/handlers/watchdog/schedule.py",
"reason": "Inherits branch-level gap: devpulse has no apps/handlers/__init__.py inspect.stack guard (manager branch). Cross-branch protection enforced at the module layer via watchdog.py _guard_caller(). Same situation as watchdog/agent.py and watchdog/timer.py."
},
{
"standard": "json_structure",
"file": "apps/handlers/watchdog/schedule.py",
"reason": "Devpulse is a manager branch with no json_handler. Schedule handler logs through prax system_logger. Same situation as watchdog/agent.py and watchdog/timer.py."
},
{
"standard": "encapsulation",
"file": "apps/handlers/watchdog/registry.py",
"reason": "Inherits branch-level gap: devpulse has no apps/handlers/__init__.py inspect.stack guard (manager branch). Cross-branch protection enforced at the module layer via watchdog.py _guard_caller(). Same situation as watchdog/agent.py, timer.py, schedule.py."
},
{
"standard": "json_structure",
"file": "apps/handlers/watchdog/registry.py",
"reason": "Devpulse is a manager branch with no json_handler. Registry logs through prax system_logger. Same situation as watchdog/agent.py, timer.py, schedule.py."
},
{
"standard": "encapsulation",
"file": "tests/test_watchdog_module.py",
"reason": "Test file imports aipass.devpulse.apps.modules.watchdog for direct router testing (same-branch import). Encapsulation check pattern-matches on 'from aipass.*.apps.*' shape regardless of same- vs cross-branch. Inherits branch-level gap (manager branch, no apps/handlers/__init__.py inspect.stack guard)."
},
{
"standard": "meta",
"file": "apps/modules/feedback.py",
"reason": "META block present (lines 1-4) but checker format mismatch — same issue as apps/devpulse.py."
},
{
"standard": "encapsulation",
"file": "apps/modules/feedback.py",
"reason": "Same-branch import from devpulse handlers. Manager branch has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "json_structure",
"file": "apps/modules/feedback.py",
"reason": "Devpulse is a manager branch with no json_handler. Feedback module logs through prax system_logger."
},
{
"standard": "introspection",
"file": "apps/modules/feedback.py",
"reason": "Feedback is a subcommand module routed via devpulse.py handle_command(), not a standalone entry point. Introspection is handled at the devpulse.py level."
},
{
"standard": "encapsulation",
"file": "apps/handlers/feedback/compose.py",
"reason": "Same-branch handler import. Manager branch has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "json_structure",
"file": "apps/handlers/feedback/compose.py",
"reason": "Devpulse is a manager branch with no json_handler. Compose handler logs through prax system_logger."
"reason": "No 'manager' citizen_class template in spawn. Devpulse is the only manager branch."
},
{
"standard": "cli",
"file": "apps/handlers/feedback/compose.py",
"reason": "Feedback handlers use console.print for Rich-formatted output routed through err_console. This is the intended output path for a manager-branch utility."
},
{
"standard": "encapsulation",
"file": "apps/handlers/feedback/inbox.py",
"reason": "Same-branch handler import. Manager branch has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "json_structure",
"file": "apps/handlers/feedback/inbox.py",
"reason": "Devpulse is a manager branch with no json_handler. Inbox handler uses Rich console output."
"reason": "Feedback handler — console.print() IS the output. Display functions are the handler's purpose."
},
{
"standard": "cli",
"file": "apps/handlers/feedback/inbox.py",
"reason": "Feedback handlers use console.print for Rich-formatted output routed through err_console. This is the intended output path for a manager-branch utility."
},
{
"standard": "unused_function",
"file": "apps/handlers/watchdog/schedule.py",
"reason": "format_wait() has 5 passing tests. Public utility for schedule display — not called in production yet but tested and documented for upcoming schedule status output."
},
{
"standard": "encapsulation",
"file": "apps/handlers/feedback/storage.py",
"reason": "Same-branch handler import. Manager branch has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "json_structure",
"file": "apps/handlers/feedback/storage.py",
"reason": "Devpulse is a manager branch with no json_handler. Storage handler uses json directly for local feedback inbox persistence."
"reason": "Feedback handler — console.print() IS the output. Display functions are the handler's purpose."
},
{
"standard": "naming",
"file": "apps/handlers/watchdog/registry.py",
"reason": "_FileLock uses Python's _PrivateClass convention (underscore prefix + PascalCase). Checker pattern ^[A-Z] doesn't account for private class prefix."
"reason": "_FileLock uses Python _PrivateClass convention (underscore prefix + PascalCase)."
},
{
"standard": "imports",
"file": "apps/handlers/feedback/compose.py",
"reason": "_AIPASS_ROOT used as fallback for ai_mail inbox resolution when CWD-based lookup fails. Feedback channel only works between dev-install branches — pip users don't have cross-branch communication. Real fix is registry-based path resolution (same class as DPLAN-0149 pip install gaps)."
"reason": "_AIPASS_ROOT fallback for ai_mail inbox resolution. Known gap — needs registry-based path resolution."
},
{
"standard": "encapsulation",
"file": "tests/test_git_gate.py",
"reason": "Test imports git_gate.py from ~/.claude/hooks/ via importlib.util.spec_from_file_location. git_gate is a user-level hook (not a branch module), so no aipass package path exists. No cross-branch handler import — this is outside the branch tree entirely."
},
{
"standard": "encapsulation",
"file": "tools/spot_check.py",
"reason": "Devpulse tool script (not a handler). Manager branch has no apps/handlers/__init__.py inspect.stack guard."
},
{
"standard": "debug_print",
"file": "tools/spot_check.py",
"reason": "Standalone CLI tool — print() is the intended user-facing output method, not debug noise."
"standard": "unused_function",
"file": "apps/handlers/watchdog/schedule.py",
"reason": "format_wait() has 5 passing tests. Public utility for schedule display, not called in production yet."
},
{
"standard": "windows_compat",
"file": "apps/handlers/watchdog/registry.py",
"lines": [138, 149],
"reason": "fcntl imports at L138 and L149 are guarded by early return at L137 (if sys.platform == win32: return self) and runtime None check (if self._fh is not None). Windows never reaches these lines."
"reason": "fcntl imports guarded by early return (sys.platform == win32) at L137. Windows never reaches these lines."
},
{
"standard": "encapsulation",
"file": "tests/test_git_gate.py",
"reason": "Test imports git_gate.py from ~/.claude/hooks/ via importlib — external hook, not a branch module."
},
{
"standard": "encapsulation",
"file": "tools/spot_check.py",
"reason": "Standalone diagnostic tool script, not a handler."
},
{
"standard": "debug_print",
"file": "tools/spot_check.py",
"reason": "Standalone CLI tool — print() is the intended output method."
}
],
"notes": {
+8 -7
View File
@@ -2,9 +2,9 @@
# DevPulse
> Orchestration hub for AIPass. Plans, coordinates, dispatches. Never the builder.
> Orchestration hub for AIPass. Plans, coordinates, dispatches, and builds its own modules.
DevPulse is the central coordination branch. It does not ship features of its own — it works with the user on design, dispatches real work to branch specialists, tracks plans and memory, and keeps the system moving. If a task belongs to another branch, DevPulse emails that branch and waits for the reply.
DevPulse is the user's primary AI collaborator. It designs, plans, debugs, and coordinates the 12 branches. Builds its own modules directly (watchdog, feedback, json_handler). Ventures into other branches to investigate, debug, and fix small bugs. Delegates heavy multi-file builds to sub-agents via dispatch.
## Start here
@@ -14,7 +14,6 @@ DevPulse is the central coordination branch. It does not ship features of its ow
| What's happening right now | [STATUS.local.md](STATUS.local.md) |
| Identity, memory, session history | [`.trinity/`](.trinity/) |
| Diagnostic scanners | [`tools/`](tools/) |
| Branch health audits | [`branch_audits _only/`](branch_audits%20_only/) |
| Active plans | `drone @flow list open` |
## Invoke
@@ -28,7 +27,7 @@ Say "hi" and DevPulse picks up where the last session left off.
## Role in one line
Designer, orchestrator, and light builder — the user's primary AI collaborator. Builds its own things directly (modules, plans, memories, design docs). Ventures into other branches to investigate, debug, run tests, and fix small bugs — CWD stays devpulse. Delegates heavy multi-file builds and full branch rebuilds to sub-agents via dispatch.
The user's primary AI collaborator — designs, orchestrates, and builds own modules. Delegates heavy multi-file builds to sub-agents via dispatch.
## Architecture
@@ -44,8 +43,9 @@ src/aipass/devpulse/
│ │ └── watchdog.py # Directed wake system routing
│ └── handlers/
│ ├── feedback/ # Feedback inbox, compose, storage
│ ├── json/ # JSON operation logging (json_handler)
│ └── watchdog/ # Agent, timer, schedule, registry handlers
├── tests/ # 130+ tests (watchdog + feedback)
├── tests/ # 236 tests (watchdog + feedback + json + devpulse)
├── docs/ # Transition notes, research
├── docs.local/ # Local-only docs (gitignored)
└── STATUS.local.md # Current work beacon
@@ -56,7 +56,8 @@ src/aipass/devpulse/
Devpulse commands are accessed via `drone @devpulse <command>`:
- `watchdog agent @target` — monitor a dispatched agent until it finishes
- `watchdog timer <seconds>` — simple countdown timer
- `watchdog timer <duration>` — wake after duration (5m, 30s, 2h, 1h30m)
- `watchdog timer start/stop <name>` — named duration tracking
- `watchdog schedule <HH:MM>` — wait until a specific time
- `watchdog status` — show active watchdogs
- `watchdog cancel <id>` — cancel a running watchdog
@@ -75,7 +76,7 @@ drone (routing), prax (logging), cli (display), ai_mail (dispatch), seedgo (audi
### Provides To
All branches via dispatch orchestration. Watchdog monitoring for any dispatched agent. Feedback channel for cross-project communication.
*Last Updated: 2026-04-24*
*Last Updated: 2026-05-15*
---
+29 -10
View File
@@ -1,3 +1,11 @@
# =================== AIPass ====================
# Name: devpulse.py
# Description: Entry point for devpulse branch — auto-discovers modules
# Version: 1.0.0
# Created: 2026-03-07
# Modified: 2026-05-15
# =============================================
"""
DEVPULSE Branch - Main Orchestrator
@@ -7,22 +15,15 @@ Auto-discovery architecture:
- No manual imports or routing needed
"""
# META
# module: devpulse
# description: Orchestration hub — coordinates via dispatch + agents
# citizen_class: manager
# END META
import sys
import importlib
from pathlib import Path
from typing import Any
from aipass.prax import logger
from aipass.cli.apps.modules import err_console
from rich.console import Console
console = Console(stderr=True)
console = err_console
# =============================================================================
# MODULE DISCOVERY
@@ -81,6 +82,24 @@ def print_introspection():
console.print(f" {name:20} {desc}")
def print_help():
"""Print CLI help — usage instructions and available commands."""
modules = discover_modules()
console.print("[bold cyan]DEVPULSE[/bold cyan] — Usage")
console.print()
console.print(" drone @devpulse <command> [args...]")
console.print()
console.print("[bold]COMMANDS:[/bold]")
for module in modules:
name = module.__name__.split(".")[-1]
desc = (module.__doc__ or "").strip().split("\n")[0] if module.__doc__ else "No description"
console.print(f" {name:20} {desc}")
console.print()
console.print("[bold]FLAGS:[/bold]")
console.print(" --help, -h Show this help message")
console.print(" --version, -V Show version")
def route_command(command: str, args: list[str], modules: list[Any]) -> bool:
"""Route command to appropriate module."""
for module in modules:
@@ -111,7 +130,7 @@ def _handle_command(command: str, args: list) -> bool:
modules = discover_modules()
if command in ["--help", "-h", "help"]:
print_introspection()
print_help()
return True
if command in ["--version", "-V"]:
@@ -26,6 +26,7 @@ from aipass.devpulse.apps.handlers.feedback.storage import (
)
from aipass.cli.apps.modules import err_console
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -81,6 +82,7 @@ def send_feedback(from_branch: str, subject: str, body: str, ai_mail_path: str =
Returns:
str: The generated message ID.
"""
json_handler.log_operation("send_feedback", {"from_branch": from_branch, "subject": subject})
data = load_inbox()
msg_id = generate_id()
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
@@ -18,6 +18,7 @@ from rich.table import Table
from aipass.devpulse.apps.handlers.feedback.storage import load_inbox, save_inbox
from aipass.cli.apps.modules import err_console
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -27,6 +28,7 @@ def list_messages() -> None:
Shows id, from, subject, date, and read status for each message.
"""
json_handler.log_operation("list_messages")
data = load_inbox()
messages = data.get("messages", [])
@@ -18,6 +18,7 @@ import secrets
from pathlib import Path
from aipass.prax import logger
from aipass.devpulse.apps.handlers.json import json_handler
# devpulse/ root (three levels up from this file: handlers/feedback/storage.py -> apps/ -> devpulse/)
_DEVPULSE_ROOT = Path(__file__).resolve().parents[3]
@@ -43,6 +44,7 @@ def load_inbox() -> dict:
Returns:
dict: The inbox data with mailbox, total_messages, unread_count, and messages.
"""
json_handler.log_operation("load_inbox")
_ensure_dir()
inbox_path = get_inbox_path()
@@ -0,0 +1,199 @@
# =================== AIPass ====================
# Name: json_handler.py
# Description: JSON auto-creating handler for devpulse data files
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""JSON auto-creating handler for devpulse data files.
Provides log_operation() for structured operation logging and
ensure_json_file() for auto-creating branch-scoped JSON files.
"""
from __future__ import annotations
import inspect
import json
import os
import tempfile
from datetime import datetime
from pathlib import Path
from typing import Any
from aipass.prax import logger
_BRANCH_ROOT: Path = Path(__file__).resolve().parents[3]
_BRANCH_NAME: str = _BRANCH_ROOT.name
JSON_DIR: Path = _BRANCH_ROOT / f"{_BRANCH_NAME}_json"
_JSON_TYPES: tuple[str, ...] = ("config", "data", "log")
def _today() -> str:
"""Return today's date as ISO string."""
return datetime.now().date().isoformat()
def _get_caller_module_name() -> str:
stack = inspect.stack()
if len(stack) > 2:
caller_path = Path(stack[2].filename)
module_name = caller_path.stem
if module_name and not module_name.startswith("_"):
return module_name
return "unknown"
def _atomic_write_json(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fd, tmp_path = tempfile.mkstemp(dir=str(path.parent), suffix=".tmp", prefix=".json_")
try:
with os.fdopen(fd, "w", encoding="utf-8") as fh:
json.dump(data, fh, indent=2, ensure_ascii=False)
os.replace(tmp_path, str(path))
except BaseException as exc:
logger.warning("_atomic_write_json: failed for %s: %s", path, exc)
try:
os.unlink(tmp_path)
except OSError as cleanup_exc:
logger.warning("_atomic_write_json: cleanup failed for %s: %s", tmp_path, cleanup_exc)
raise
def _default_config(module_name: str) -> dict[str, Any]:
today = _today()
return {
"module_name": module_name,
"version": "1.0.0",
"config": {"max_log_entries": 100},
"created": today,
"last_updated": today,
}
def _default_data(module_name: str) -> dict[str, Any]: # noqa: ARG001
today = _today()
return {"created": today, "last_updated": today}
def _default_log(module_name: str) -> list[Any]: # noqa: ARG001
return []
_DEFAULTS: dict[str, Any] = {
"config": _default_config,
"data": _default_data,
"log": _default_log,
}
def validate_json_structure(data: Any, json_type: str) -> bool:
"""Check that data matches expected shape for json_type."""
if json_type == "config":
if not isinstance(data, dict):
return False
return all(key in data for key in ("module_name", "version", "config"))
if json_type == "data":
if not isinstance(data, dict):
return False
return all(key in data for key in ("created", "last_updated"))
if json_type == "log":
return isinstance(data, list)
return False
def get_json_path(module_name: str, json_type: str) -> Path:
"""Return filesystem path for a module's JSON file."""
return JSON_DIR / f"{module_name}_{json_type}.json"
def ensure_json_exists(module_name: str, json_type: str) -> bool:
"""Ensure a single JSON file exists, creating with defaults if missing."""
JSON_DIR.mkdir(parents=True, exist_ok=True)
json_path = get_json_path(module_name, json_type)
if json_path.exists():
try:
if json_path.stat().st_size == 0:
logger.warning("ensure_json_exists: empty file at %s, regenerating", json_path)
else:
with open(json_path, encoding="utf-8") as fh:
data = json.load(fh)
if validate_json_structure(data, json_type):
return True
except Exception as exc: # noqa: BLE001
logger.warning("ensure_json_exists: failed to read %s, regenerating: %s", json_path, exc)
factory = _DEFAULTS.get(json_type)
if factory is None:
raise ValueError(f"Unknown json_type: {json_type!r}")
default = factory(module_name)
_atomic_write_json(json_path, default)
return True
def ensure_module_jsons(module_name: str) -> bool:
"""Ensure all three JSON files (config, data, log) exist for a module."""
for json_type in _JSON_TYPES:
ensure_json_exists(module_name, json_type)
return True
def load_json(module_name: str, json_type: str) -> Any | None:
"""Load a module's JSON file, auto-creating if missing."""
if not ensure_json_exists(module_name, json_type):
return None
json_path = get_json_path(module_name, json_type)
try:
if json_path.stat().st_size == 0:
factory = _DEFAULTS.get(json_type)
return factory(module_name) if factory else None
with open(json_path, encoding="utf-8") as fh:
return json.load(fh)
except (json.JSONDecodeError, OSError) as exc:
logger.warning("load_json: failed to read %s: %s", json_path, exc)
factory = _DEFAULTS.get(json_type)
return factory(module_name) if factory else None
def save_json(module_name: str, json_type: str, data: Any) -> bool:
"""Write data to a module's JSON file after validation."""
if not validate_json_structure(data, json_type):
raise ValueError(f"Invalid structure for {json_type} JSON")
if json_type == "data" and isinstance(data, dict):
data["last_updated"] = _today()
json_path = get_json_path(module_name, json_type)
_atomic_write_json(json_path, data)
return True
def log_operation(
operation: str,
data: dict[str, Any] | None = None,
module_name: str | None = None,
) -> bool:
"""Append an entry to a module's operation log with FIFO rotation."""
if module_name is None:
module_name = _get_caller_module_name()
try:
ensure_module_jsons(module_name)
config = load_json(module_name, "config")
max_entries = 100
if config and "config" in config:
max_entries = config["config"].get("max_log_entries", 100)
log = load_json(module_name, "log")
if log is None:
log = []
entry: dict[str, Any] = {
"timestamp": datetime.now().isoformat(),
"operation": operation,
}
if data:
entry["data"] = data
log.append(entry)
if len(log) > max_entries:
log = log[-max_entries:]
return save_json(module_name, "log", log)
except Exception as exc:
logger.warning("log_operation: failed for %s/%s: %s", module_name, operation, exc)
return False
@@ -35,6 +35,7 @@ from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.devpulse.apps.handlers.watchdog import registry as _registry
from aipass.devpulse.apps.handlers.json import json_handler
def _stderr(msg: str) -> None:
@@ -404,7 +405,7 @@ def watch_agent(
elapsed_int = int(time.monotonic() - started_at)
state, reason, exit_code = _classify_exit(branch_path, lock_existed=True, dispatch_ts=dispatch_ts)
logger.info("[watchdog.agent] wake agent_id=%s state=%s elapsed=%s", agent_id, state, elapsed_int)
return {
result = {
"woke": True,
"reason": reason,
"elapsed": elapsed_int,
@@ -413,6 +414,8 @@ def watch_agent(
"agent_id": agent_id,
"handle": handle,
}
json_handler.log_operation("watch_agent", {"agent_id": agent_id, "state": result.get("agent_state")})
return result
if isinstance(initial_pid, int) and not _pid_alive(initial_pid):
_stderr(
@@ -50,6 +50,7 @@ from datetime import datetime
from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.devpulse.apps.handlers.json import json_handler
_STORAGE_FILENAME = "watchdog_active.json"
@@ -230,6 +231,7 @@ def register(
store["watches"].append(entry)
_atomic_write_unlocked(path, store)
json_handler.log_operation("register_watch", {"watch_type": watch_type})
logger.info("[watchdog.registry] register type=%s handle=%s pid=%s", watch_type, handle, entry["pid"])
return handle
@@ -32,6 +32,7 @@ from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.devpulse.apps.handlers.watchdog import registry as _registry
from aipass.devpulse.apps.handlers.watchdog.timer import parse_duration
from aipass.devpulse.apps.handlers.json import json_handler
# Chunk size keeps long sleeps interruptible and makes tests with injected
@@ -163,6 +164,7 @@ def wake_at(
``command_stderr``. Command fields are ``None`` when no command
was requested.
"""
json_handler.log_operation("wake_at", {"time_str": time_str, "command": command})
clock = now_fn if now_fn is not None else datetime.now
start = clock()
target = parse_schedule(time_str, now=start)
@@ -35,6 +35,7 @@ from pathlib import Path
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.devpulse.apps.handlers.watchdog import registry as _registry
from aipass.devpulse.apps.handlers.json import json_handler
_DURATION_TOKEN_RE = re.compile(r"(\d+)([smh])")
@@ -180,6 +181,7 @@ def wake_in(duration: str) -> dict:
orchestrators with a console attached can see progress. Short timers
stay silent so test runs aren't chatty.
"""
json_handler.log_operation("wake_in", {"duration": duration})
total_seconds = parse_duration(duration)
started_at = time.monotonic()
logger.info("[watchdog.timer] wake_in duration=%s total=%ss", duration, total_seconds)
+21 -5
View File
@@ -1,7 +1,10 @@
# META
# module: devpulse.feedback
# description: Command routing module for feedback mailbox
# END META
# =================== AIPass ====================
# Name: feedback.py
# Description: Feedback Module — command routing for devpulse feedback mailbox
# Version: 1.0.0
# Created: 2026-04-11
# Modified: 2026-05-15
# =============================================
"""
Feedback Module — command routing for devpulse's personal feedback mailbox.
@@ -25,6 +28,7 @@ from aipass.devpulse.apps.handlers.feedback.compose import (
from aipass.prax import logger
from aipass.cli.apps.modules import err_console
from aipass.devpulse.apps.handlers.json import json_handler
console = err_console
@@ -43,6 +47,17 @@ HELP_TEXT = """\
"""
def print_introspection() -> None:
"""Display module introspection info."""
console.print()
console.print("feedback Module")
console.print("DevPulse personal feedback mailbox. Receives cross-project")
console.print("feedback messages from any agent via drone routing.")
console.print()
console.print("Subcommands: inbox, view, reply, send, clear")
console.print()
def handle_command(command: str, args: list[str]) -> bool:
"""Route feedback commands to handler functions.
@@ -58,14 +73,15 @@ def handle_command(command: str, args: list[str]) -> bool:
if command != "feedback":
return False
# No subcommand — show summary
if not args:
print_introspection()
summary = get_summary()
console.print(f"[bold cyan]Feedback:[/bold cyan] {summary}")
return True
subcommand = args[0]
sub_args = args[1:]
json_handler.log_operation("feedback_command", {"subcommand": subcommand})
if subcommand in ("--help", "-h", "help"):
console.print(HELP_TEXT)
@@ -30,6 +30,7 @@ from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, error, warning
from aipass.devpulse.apps.handlers.json import json_handler
_VALID_SUBCOMMANDS = ["agent", "timer", "schedule", "status", "cancel", "list"]
_DEFAULT_AGENT_TIMEOUT = 600
@@ -154,6 +155,7 @@ def handle_command(command: str, args: List[str]) -> bool:
return True
logger.info("[watchdog] subcommand=%s args=%s", subcommand, sub_args)
json_handler.log_operation("watchdog_command", {"subcommand": subcommand, "args": sub_args})
if subcommand == "agent":
return _handle_agent(sub_args)
+23 -16
View File
@@ -1,19 +1,15 @@
# ===================AIPASS====================
# META DATA HEADER
# Name: tests/conftest.py
# Date: 2025-11-08
# Version: 1.0.0
# Category: cortex/tests
#
# CHANGELOG (Max 5 entries):
# - v1.0.0 (2025-11-08): Initial implementation - Shared pytest fixtures
#
# CODE STANDARDS:
# - Error handling: Use error handler system (apps/handlers/error/)
# =================== AIPass ====================
# Name: conftest.py
# Description: Shared pytest fixtures for devpulse tests
# Version: 1.1.0
# Created: 2025-11-08
# Modified: 2026-05-15
# =============================================
"""Shared pytest fixtures for cortex tests"""
from unittest.mock import patch
import pytest
import shutil
import tempfile
@@ -32,8 +28,19 @@ def temp_test_dir() -> Generator[Path, None, None]:
@pytest.fixture
def sample_test_data() -> dict:
"""Provides sample test data
Customize this fixture for your module's needs
"""
"""Provides sample test data."""
return {"test_key": "test_value", "sample_data": "example"}
@pytest.fixture
def mock_logger():
"""Mock the prax logger to suppress output during tests."""
with patch("aipass.prax.logger") as mock_log:
yield mock_log
@pytest.fixture
def mock_json_handler():
"""Mock json_handler to prevent filesystem writes during tests."""
with patch("aipass.devpulse.apps.handlers.json.json_handler.log_operation") as mock_json:
yield mock_json
+133
View File
@@ -0,0 +1,133 @@
# =================== AIPass ====================
# Name: test_devpulse.py
# Description: Tests for devpulse.py CLI routing, introspection, and resilience
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
"""Tests for devpulse.py — entry point CLI routing and module discovery."""
import importlib
from unittest.mock import patch, MagicMock
from aipass.devpulse.apps import devpulse as devpulse_module
class TestCLIRouting:
"""CLI routing through _handle_command and main()."""
def test_help_flag(self):
"""--help flag returns True."""
result = devpulse_module._handle_command("--help", [])
assert result is True
def test_short_help(self):
"""-h flag returns True."""
result = devpulse_module._handle_command("-h", [])
assert result is True
def test_help_word(self):
"""help word returns True."""
result = devpulse_module._handle_command("help", [])
assert result is True
def test_version_flag(self):
"""--version flag returns True."""
result = devpulse_module._handle_command("--version", [])
assert result is True
def test_version_short(self):
"""-V flag returns True."""
result = devpulse_module._handle_command("-V", [])
assert result is True
def test_unknown_command_returns_false(self):
"""Unrecognized command returns False."""
result = devpulse_module._handle_command("nonexistent_unknown_command", [])
assert result is False
@patch.object(devpulse_module, "print_help")
def test_print_help_called_on_help_flag(self, mock_print_help):
"""--help invokes print_help."""
devpulse_module._handle_command("--help", [])
mock_print_help.assert_called_once()
@patch.object(devpulse_module, "print_introspection")
def test_no_args_triggers_print_introspection(self, mock_introspection):
"""No args triggers print_introspection via main()."""
with patch("sys.argv", ["devpulse"]):
devpulse_module.main()
mock_introspection.assert_called_once()
@patch.object(devpulse_module, "print_introspection")
def test_print_introspection_output(self, mock_introspection):
"""main() returns 0 when print_introspection runs."""
mock_introspection.return_value = None
with patch("sys.argv", ["devpulse"]):
result = devpulse_module.main()
assert result == 0
class TestModuleDiscovery:
"""discover_modules() finds modules with handle_command."""
def test_discover_modules_returns_list(self):
"""Returns a list of discovered modules."""
result = devpulse_module.discover_modules()
assert isinstance(result, list)
def test_discovered_modules_have_handle_command(self):
"""Each discovered module exposes handle_command."""
modules = devpulse_module.discover_modules()
for mod in modules:
assert hasattr(mod, "handle_command")
def test_reimport_after_mock(self):
"""Verify module reimport picks up mocked state."""
devpulse_module.discover_modules()
with patch.object(devpulse_module, "MODULES_DIR", devpulse_module.Path("/nonexistent")):
importlib.reload(devpulse_module)
reloaded = devpulse_module.discover_modules()
importlib.reload(devpulse_module)
assert isinstance(reloaded, list)
class TestErrorResilience:
"""Graceful handling of edge cases."""
def test_route_command_catches_module_errors(self):
"""Module exceptions are caught, returns False."""
bad_module = MagicMock(__name__="bad_module")
bad_module.handle_command.side_effect = Exception("boom")
result = devpulse_module.route_command("test", [], [bad_module])
assert result is False
def test_empty_file_modules_dir(self, tmp_path):
"""discover_modules handles empty_file in modules directory."""
empty = tmp_path / "empty.py"
empty.write_text("")
with patch.object(devpulse_module, "MODULES_DIR", tmp_path):
result = devpulse_module.discover_modules()
assert isinstance(result, list)
def test_handle_command_with_empty_args(self):
"""--help with empty args list succeeds."""
result = devpulse_module.handle_command("--help", [])
assert result is True
class TestHandleCommandGuard:
"""handle_command cross-branch security guard."""
def test_handle_command_returns_bool(self):
"""handle_command always returns a bool."""
result = devpulse_module.handle_command("--help", [])
assert isinstance(result, bool)
assert result is True
def test_handle_command_unknown_returns_false(self):
"""Unknown commands return False through the guard."""
result = devpulse_module.handle_command("bogus_invalid_command", [])
assert result is False
@@ -6,10 +6,11 @@
# Modified: 2026-05-12
# =============================================
"""Dev branch PR handler — push dev and create PR to main."""
"""Branch PR handlers — generic and dev-specific."""
from __future__ import annotations
import re
import subprocess
from aipass.prax import logger
@@ -17,6 +18,101 @@ from aipass.drone.apps.handlers.json import json_handler
from aipass.drone.apps.handlers.git.lock_handler import find_repo_root
def _slugify(text: str) -> str:
"""Convert text to a branch-safe slug."""
slug = text.lower().strip().replace(" ", "-")
slug = re.sub(r"[^a-z0-9-]", "", slug)
slug = re.sub(r"-{2,}", "-", slug).strip("-")
return slug[:40]
def create_branch_pr(description: str, target_branch: str = "main") -> dict:
"""Push current branch and create a PR to target.
Works from any branch. Detects HEAD, pushes to origin, creates PR via gh CLI.
Returns:
Dict with success, message, and pr_url keys.
"""
repo_root = find_repo_root()
try:
head = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("Failed to detect current branch: %s", exc)
return {"success": False, "message": f"Failed to detect current branch: {exc}", "pr_url": ""}
pr_branch = head.stdout.strip()
if pr_branch == target_branch:
pr_branch = _slugify(description)
if not pr_branch:
return {"success": False, "message": "Description required to generate branch name.", "pr_url": ""}
push_refspec = f"{target_branch}:{pr_branch}"
else:
push_refspec = pr_branch
try:
push = subprocess.run(
["git", "push", "-u", "origin", push_refspec],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except (OSError, subprocess.SubprocessError) as exc:
logger.error("git push failed: %s", exc)
return {"success": False, "message": f"Push failed: {exc}", "pr_url": ""}
if push.returncode != 0:
return {"success": False, "message": f"Push failed: {push.stderr.strip()}", "pr_url": ""}
try:
pr = subprocess.run(
["gh", "pr", "create", "--head", pr_branch, "--base", target_branch, "--title", description, "--body", ""],
capture_output=True,
text=True,
cwd=str(repo_root),
)
except FileNotFoundError as exc:
logger.warning("gh CLI not found: %s", exc)
return {"success": False, "message": "gh CLI not found. Install: https://cli.github.com/", "pr_url": ""}
except (OSError, subprocess.SubprocessError) as exc:
logger.error("gh pr create failed: %s", exc)
return {"success": False, "message": f"PR creation failed: {exc}", "pr_url": ""}
if pr.returncode != 0:
stderr = pr.stderr.strip()
if "already exists" in stderr:
existing_url = ""
for line in stderr.splitlines():
if "github.com" in line:
existing_url = line.strip()
break
msg = (
f"Pushed to {pr_branch}. PR already open: {existing_url}"
if existing_url
else f"Pushed to {pr_branch}. PR already open."
)
json_handler.log_operation(
"branch_pr_push_existing", {"pr_url": existing_url, "branch": pr_branch, "description": description}
)
return {"success": True, "message": msg, "pr_url": existing_url}
return {"success": False, "message": f"PR creation failed: {stderr}", "pr_url": ""}
pr_url = pr.stdout.strip()
json_handler.log_operation(
"create_branch_pr", {"pr_url": pr_url, "branch": pr_branch, "target": target_branch, "description": description}
)
logger.info("PR created from %s: %s", pr_branch, pr_url)
return {"success": True, "message": f"PR created: {pr_url}", "pr_url": pr_url}
def create_dev_pr(description: str) -> dict:
"""Push dev branch and create a PR to main.
+18 -17
View File
@@ -55,7 +55,6 @@ _COMMANDS = (
"unlock",
"dev-pr",
"delete-branch",
"system-pr",
"merge",
"smart-sync",
"fix",
@@ -153,8 +152,6 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
return _handle_sync(args)
if command == "unlock":
return _handle_unlock(args)
if command == "system-pr":
return _handle_system_pr(args, caller)
if command == "merge":
return _handle_merge(args, caller)
if command == "smart-sync":
@@ -162,7 +159,7 @@ def handle_command(command: str | None = None, args: list[str] | None = None) ->
if command == "fix":
return _handle_fix(args, caller)
if command == "pr":
return {"stdout": "", "stderr": "Agent PRs are deprecated.", "exit_code": 1}
return _handle_pr(args)
available = ", ".join(_COMMANDS)
return {
@@ -215,6 +212,21 @@ def _handle_branches() -> dict:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
def _handle_pr(args: list[str]) -> dict:
"""Handle the pr subcommand — push current branch and create PR to main."""
if not args:
return {
"stdout": "",
"stderr": "Usage: drone @git pr <description>",
"exit_code": 1,
}
description = " ".join(args)
result = dev_pr_handler.create_branch_pr(description)
if result["success"]:
return {"stdout": result["message"], "stderr": "", "exit_code": 0}
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _handle_dev_pr(args: list[str]) -> dict:
"""Handle the dev-pr subcommand (owner tier)."""
if not args:
@@ -244,15 +256,6 @@ def _handle_delete_branch(args: list[str]) -> dict:
return {"stdout": "", "stderr": result["message"], "exit_code": 1}
def _handle_system_pr(_args: list[str], _caller: str) -> dict:
"""Handle the system-pr subcommand — DEPRECATED."""
return {
"stdout": "",
"stderr": "system-pr is deprecated. Use: drone @git dev-pr <description>",
"exit_code": 1,
}
def _handle_merge(args: list[str], caller: str) -> dict:
"""Handle the merge subcommand (owner-tier, auth pre-checked)."""
if not args:
@@ -542,7 +545,7 @@ def get_help(command: str | None = None) -> str:
"git workflow [args] — Passthrough to gh workflow CLI [global]\n Examples: list, view <name>, run <name>\n"
)
if command == "pr":
return "git pr — DEPRECATED. Agent PRs are no longer supported. Devpulse handles git.\n"
return "git pr <description> — Push current branch and create PR to main [owner]\n"
if command == "status":
return "git status [--all] — Show git status filtered to your branch (--all for repo-wide) [global]\n"
if command == "diff":
@@ -585,8 +588,6 @@ def get_help(command: str | None = None) -> str:
)
if command == "unlock":
return "git unlock --force — Force-release the PR lock [owner]\n"
if command == "system-pr":
return "git system-pr — DEPRECATED. Use: drone @git dev-pr <description>\n"
if command == "merge":
return (
"git merge <PR#> — Merge a PR and sync local main [owner]\n"
@@ -624,13 +625,13 @@ def get_help(command: str | None = None) -> str:
"Owner (devpulse only):\n"
" commit <msg> [--all | files] Commit changes (selective or --all)\n"
" checkout <main|dev> Switch branches\n"
" pr <desc> Push current branch and create PR to main\n"
" dev-pr <desc> Push dev and create PR to main\n"
" delete-branch <name> Delete a remote branch\n"
" merge <PR#> Merge a PR\n"
" sync [--autostash] Checkout main and pull\n"
" smart-sync Fetch + rebase if behind\n"
" unlock --force Force-release the PR lock\n"
" system-pr DEPRECATED (use dev-pr)\n"
" fix [--dry-run] Fix broken git states\n"
)
@@ -35,11 +35,11 @@ GIT_ACCESS_TIERS: dict[str, dict] = {
"checkout",
"sync",
"unlock",
"system-pr",
"merge",
"smart-sync",
"fix",
"dev-pr",
"pr",
"delete-branch",
],
"allowed_callers": ["devpulse"],
@@ -121,9 +121,6 @@ def verify_git_access(command: str) -> str:
Raises:
PermissionError: If the caller is not authorized for this command.
"""
if command == "pr":
raise PermissionError("Agent PRs are deprecated. Build code, run tests, report results. Devpulse handles git.")
global_cmds = GIT_ACCESS_TIERS["global"]["commands"]
owner_tier = GIT_ACCESS_TIERS["owner"]
@@ -539,6 +539,7 @@ class MonitoringFileHandler(FileSystemEventHandler):
"""Parse Claude Code session JSONL to show agent actions.
Returns True if an event was emitted (or deduped), False on failure.
Iterates forward through new lines, emitting all distinct actions.
"""
try:
path_key = str(file_path)
@@ -559,14 +560,13 @@ class MonitoringFileHandler(FileSystemEventHandler):
if not lines:
return True
for line in reversed(lines):
for line in lines:
try:
entry = _json.loads(line)
except _json.JSONDecodeError as e:
logger.info(f"[monitor] Skipping malformed JSONL line: {e}")
continue
# Extract model for display tag
model = self._extract_model_from_entry(entry)
if model:
self._session_models[path_key] = self._shorten_model(model)
@@ -576,7 +576,7 @@ class MonitoringFileHandler(FileSystemEventHandler):
continue
if self._last_agent_action.get(path_key) == action_text:
return True
continue
self._last_agent_action[path_key] = action_text
tagged_branch = self._tag_branch_with_model(path_key, branch)
@@ -590,7 +590,6 @@ class MonitoringFileHandler(FileSystemEventHandler):
)
if self._event_queue:
self._event_queue.enqueue(evt)
return True
return True
@@ -1,5 +1,5 @@
# README Standards
**Status:** Active v1.0
**Status:** Active v1.1
**Date:** 2026-02-21
---
@@ -97,17 +97,19 @@ Auto-generation handles facts (file lists, timestamps). Humans handle meaning.
| Tool | Purpose | Location |
|------|---------|----------|
| `readme_check.py` | 6 automated checks, score >= 75% to pass | `src/aipass/seedgo/apps/standards/aipass/handlers/standards/` |
| `readme_check.py` | 8 automated checks, score >= 75% to pass | `src/aipass/seedgo/apps/standards/aipass/handlers/standards/` |
| `readme_generator.py` | Auto-populates TREE, MODULES, COMMANDS, HEADER, LAST_UPDATED | `src/aipass/seedgo/apps/standards/aipass/handlers/standards/` |
| `seedgo readme update @branch` | On-demand regeneration (Phase 4, coming soon) | CLI |
**Checks performed by `readme_check.py`:**
1. README.md exists
2. Contains auto-generated markers
2. Required sections present
3. Last Updated is within 7 days
4. Directory tree matches filesystem
5. Module list is complete
6. Required sections present
6. Command list presence
7. Test count accuracy (claimed vs actual `def test_` count, >10% drift fails)
8. Markdown link validity (relative `[text](path)` links point to existing files)
---
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: readme_check.py
# Description: README Standards Checker Handler
# Version: 1.0.0
# Version: 1.1.0
# Created: 2026-03-05
# Modified: 2026-03-05
# Modified: 2026-05-15
# =============================================
"""
@@ -18,6 +18,8 @@ Checks:
4. Directory tree accuracy (mentioned directories exist on disk)
5. Module list completeness (all modules in apps/modules/ mentioned)
6. Command list presence (commands/usage section is not empty)
7. Test count accuracy (claimed count vs actual def test_ functions)
8. Markdown link validity (relative links point to existing paths)
"""
import os
@@ -77,6 +79,8 @@ def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
"Directory tree accuracy",
"Module list completeness",
"Command list presence",
"Test count accuracy",
"Markdown link validity",
]:
checks.append({"name": name, "passed": False, "message": "Cannot check - README.md missing"})
@@ -119,6 +123,14 @@ def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
commands_check = check_command_list(lines, module_path, bypass_rules)
checks.append(commands_check)
# Check 7: Test count accuracy
test_count_check = check_test_count_accuracy(lines, branch_root, module_path, bypass_rules)
checks.append(test_count_check)
# Check 8: Markdown link validity
link_check = check_markdown_links(lines, branch_root, module_path, bypass_rules)
checks.append(link_check)
# Calculate score
passed_checks = sum(1 for c in checks if c["passed"])
total_checks = len(checks)
@@ -407,6 +419,141 @@ def check_module_list(lines: List[str], branch_root: Path, file_path: str, bypas
}
def check_test_count_accuracy(
lines: List[str], branch_root: Path, file_path: str, bypass_rules: list | None = None
) -> Dict:
"""
Check that test count claims in README match actual test function count.
Scans README for patterns like "N tests", "N test functions", etc.
Counts actual `def test_` functions in tests/ directory.
Flags when claimed count drifts >10% from actual.
"""
if is_bypassed(file_path, "readme", None, bypass_rules):
return {"name": "Test count accuracy", "passed": True, "message": "Bypassed by bypass rules"}
content = "\n".join(lines)
claimed_counts = _extract_test_counts(content)
if not claimed_counts:
return {
"name": "Test count accuracy",
"passed": True,
"message": "No test count claims found in README (skipped)",
}
tests_dir = branch_root / "tests"
if not tests_dir.exists():
return {
"name": "Test count accuracy",
"passed": True,
"message": "No tests/ directory found (skipped)",
}
actual_count = _count_test_functions(tests_dir)
max_claimed = max(claimed_counts)
if actual_count == 0:
if max_claimed > 0:
return {
"name": "Test count accuracy",
"passed": False,
"message": f"README claims {max_claimed} tests but no test functions found",
}
return {"name": "Test count accuracy", "passed": True, "message": "Both README and tests/ show 0 tests"}
drift_pct = abs(max_claimed - actual_count) / actual_count * 100
if drift_pct <= 10:
return {
"name": "Test count accuracy",
"passed": True,
"message": f"Test count claim ({max_claimed}) within 10% of actual ({actual_count})",
}
return {
"name": "Test count accuracy",
"passed": False,
"message": f"README claims {max_claimed} tests, actual count is {actual_count} ({drift_pct:.0f}% drift)",
}
def _extract_test_counts(content: str) -> List[int]:
"""Extract numeric test count claims from README content."""
counts = []
pattern = re.compile(r"\b(\d+)\s+tests?\b", re.IGNORECASE)
for match in pattern.finditer(content):
counts.append(int(match.group(1)))
return counts
def _count_test_functions(tests_dir: Path) -> int:
"""Count `def test_` functions in all test_*.py files under tests/."""
count = 0
test_func_pattern = re.compile(r"^\s*def\s+test_", re.MULTILINE)
for test_file in tests_dir.rglob("test_*.py"):
try:
source = test_file.read_text(encoding="utf-8")
count += len(test_func_pattern.findall(source))
except OSError:
logger.info("Cannot read test file %s for count", test_file)
continue
return count
def check_markdown_links(lines: List[str], branch_root: Path, file_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check that relative markdown links point to existing paths.
Parses [text](path) links where path is relative (not http/https/mailto/#).
Verifies each path exists relative to branch root.
"""
if is_bypassed(file_path, "readme", None, bypass_rules):
return {"name": "Markdown link validity", "passed": True, "message": "Bypassed by bypass rules"}
content = "\n".join(lines)
links = _extract_relative_links(content)
if not links:
return {
"name": "Markdown link validity",
"passed": True,
"message": "No relative markdown links found (skipped)",
}
dead_links = []
for link_text, link_path in links:
resolved = (branch_root / link_path).resolve()
if not resolved.exists():
dead_links.append(f"{link_path} ({link_text})")
if not dead_links:
return {
"name": "Markdown link validity",
"passed": True,
"message": f"All {len(links)} relative links verified",
}
return {
"name": "Markdown link validity",
"passed": False,
"message": f"Dead links: {', '.join(dead_links)}",
}
def _extract_relative_links(content: str) -> List[tuple]:
"""Extract relative markdown links as (text, path) tuples."""
link_pattern = re.compile(r"\[([^\]]*)\]\(([^)]+)\)")
links = []
for match in link_pattern.finditer(content):
text = match.group(1)
path = match.group(2)
if path.startswith(("http://", "https://", "mailto:", "#")):
continue
links.append((text, path))
return links
def check_command_list(lines: List[str], file_path: str, bypass_rules: list | None = None) -> Dict:
"""
Check that README has a non-empty commands/usage section.
@@ -1,9 +1,9 @@
# =================== AIPass ====================
# Name: readme_content.py
# Description: README Standards Content
# Version: 1.0.0
# Version: 1.1.0
# Created: 2026-03-05
# Modified: 2026-03-05
# Modified: 2026-05-15
# =============================================
"""
@@ -70,7 +70,27 @@ def get_readme_standards() -> str:
The Commands/Usage section must not be empty.
At minimum, list the primary commands the branch supports.
[yellow]CHECK 7 - TEST COUNT ACCURACY:[/yellow]
If README mentions test counts (e.g. "219 tests" in tree comments
or status lines), the claimed number must be within [bold white]10%[/bold white]
of the actual [dim]def test_[/dim] function count in tests/.
The highest claimed count is compared against actual.
Branches with no test claims or no tests/ directory pass by default.
Date-bumping hides this drift. A branch can update its date
every week while claiming "130 tests" when reality is 450.
[yellow]CHECK 8 - MARKDOWN LINK VALIDITY:[/yellow]
All relative markdown links [dim]\\[text](path)[/dim] must point to
existing files or directories relative to the branch root.
Skips external links (http/https/mailto) and anchor links (#).
Dead links mislead contributors navigating via README.
[yellow]SCORING:[/yellow]
6 checks, each worth ~17 points. Pass threshold: 75%.
A branch with a missing README scores 17/100 (only check 1 runs)."""
8 checks, each worth ~12.5 points. Pass threshold: 75%.
A branch with a missing README scores 12/100 (only check 1 runs)."""
@@ -51,11 +51,17 @@ def _mock_infrastructure(monkeypatch):
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
# -- bypass handler -----------------------------------------------------
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed as real_is_bypassed
bypass_pkg = MagicMock()
bypass_utils = MagicMock()
bypass_utils.is_bypassed = real_is_bypassed
bypass_pkg.utils = bypass_utils
bypass_ignore = MagicMock()
bypass_ignore.get_template_ignore_patterns = MagicMock(return_value=[])
bypass_pkg.ignore_handler = bypass_ignore
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass", bypass_pkg)
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass.utils", bypass_utils)
monkeypatch.setitem(
sys.modules,
"aipass.seedgo.apps.handlers.bypass.ignore_handler",
@@ -0,0 +1,473 @@
"""Tests for readme_check.py — Check 7 (test count accuracy) and Check 8 (markdown link validity)."""
# =================== META ====================
# Name: test_readme_content_checks.py
# Description: Unit tests for readme content accuracy checks
# Version: 1.0.0
# Created: 2026-05-15
# Modified: 2026-05-15
# =============================================
import pytest
from typing import List
from unittest.mock import MagicMock
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _lines(text: str) -> List[str]:
return text.split("\n")
# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture(autouse=True)
def _mock_infrastructure(monkeypatch):
"""Mock heavy infrastructure imports for readme_check."""
import sys
mock_logger = MagicMock()
mock_json_handler = MagicMock()
mock_json_handler.log_operation = MagicMock(return_value=True)
prax_mod = MagicMock()
prax_mod.logger = mock_logger
monkeypatch.setitem(sys.modules, "aipass.prax", prax_mod)
json_pkg = MagicMock()
json_pkg.json_handler = mock_json_handler
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json", json_pkg)
json_mod = MagicMock()
json_mod.log_operation = mock_json_handler.log_operation
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.json.json_handler", json_mod)
from aipass.seedgo.apps.handlers.bypass.utils import is_bypassed as real_is_bypassed
bypass_pkg = MagicMock()
bypass_utils = MagicMock()
bypass_utils.is_bypassed = real_is_bypassed
bypass_pkg.utils = bypass_utils
bypass_ignore = MagicMock()
bypass_ignore.get_template_ignore_patterns = MagicMock(return_value=[])
bypass_pkg.ignore_handler = bypass_ignore
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass", bypass_pkg)
monkeypatch.setitem(sys.modules, "aipass.seedgo.apps.handlers.bypass.utils", bypass_utils)
monkeypatch.setitem(
sys.modules,
"aipass.seedgo.apps.handlers.bypass.ignore_handler",
bypass_ignore,
)
monkeypatch.delitem(
sys.modules,
"aipass.seedgo.apps.handlers.aipass_standards.readme_check",
raising=False,
)
# ===========================================================================
# 7. check_test_count_accuracy
# ===========================================================================
def test_test_count_no_claims():
"""No test count claims in README passes (skipped)."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
from pathlib import Path
lines = _lines("# Branch\n\nSome content without test counts.\n")
result = check_test_count_accuracy(lines, Path("/nonexistent"), "fake.py")
assert result["passed"] is True
assert "skipped" in result["message"].lower()
def test_test_count_no_tests_dir(tmp_path):
"""Test count claim with no tests/ directory passes (skipped)."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
lines = _lines("├── tests/ # 50 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is True
assert "skipped" in result["message"].lower()
def test_test_count_accurate(tmp_path):
"""Claimed count within 10% of actual passes."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "test_one.py").write_text(
"def test_a(): pass\ndef test_b(): pass\ndef test_c(): pass\n"
"def test_d(): pass\ndef test_e(): pass\ndef test_f(): pass\n"
"def test_g(): pass\ndef test_h(): pass\ndef test_i(): pass\n"
"def test_j(): pass\n",
encoding="utf-8",
)
lines = _lines("├── tests/ # 10 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is True
assert "within 10%" in result["message"]
def test_test_count_drift_over_10_pct(tmp_path):
"""Claimed count drifting >10% from actual fails."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "test_one.py").write_text(
"def test_a(): pass\ndef test_b(): pass\ndef test_c(): pass\n"
"def test_d(): pass\ndef test_e(): pass\ndef test_f(): pass\n"
"def test_g(): pass\ndef test_h(): pass\ndef test_i(): pass\n"
"def test_j(): pass\n",
encoding="utf-8",
)
lines = _lines("├── tests/ # 50 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is False
assert "drift" in result["message"].lower()
def test_test_count_claims_zero_actual_nonzero(tmp_path):
"""README claims tests but 0 actual functions found fails."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "test_empty.py").write_text("# no test functions\n", encoding="utf-8")
lines = _lines("├── tests/ # 30 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is False
def test_test_count_uses_max_claimed(tmp_path):
"""When multiple counts claimed, uses the highest for comparison."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
funcs = "\n".join(f"def test_{i}(): pass" for i in range(100))
(tests_dir / "test_one.py").write_text(funcs, encoding="utf-8")
lines = _lines("├── tests/ # 100 tests across 5 files\n│ ├── test_a.py # 20 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is True
def test_test_count_bypassed():
"""Bypassed standard passes immediately."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
from pathlib import Path
bypass_rules = [{"file": "fake.py", "standard": "readme", "reason": "test"}]
lines = _lines("├── tests/ # 999 tests\n")
result = check_test_count_accuracy(lines, Path("/tmp"), "fake.py", bypass_rules)
assert result["passed"] is True
def test_test_count_rglob_nested(tmp_path):
"""Counts test functions in nested test subdirectories."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_test_count_accuracy,
)
tests_dir = tmp_path / "tests"
sub_dir = tests_dir / "subdir"
sub_dir.mkdir(parents=True)
(tests_dir / "test_top.py").write_text("def test_a(): pass\n", encoding="utf-8")
(sub_dir / "test_nested.py").write_text("def test_b(): pass\n", encoding="utf-8")
lines = _lines("├── tests/ # 2 tests\n")
result = check_test_count_accuracy(lines, tmp_path, "fake.py")
assert result["passed"] is True
# ===========================================================================
# _extract_test_counts
# ===========================================================================
def test_extract_test_counts_various_patterns():
"""Extracts counts from various README patterns."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_test_counts,
)
content = "├── tests/ # 219 tests (watchdog + feedback)\n**Tests:** 219 tests passing\n"
counts = _extract_test_counts(content)
assert 219 in counts
assert len(counts) == 2
def test_extract_test_counts_singular():
"""Matches singular 'test' as well as plural."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_test_counts,
)
counts = _extract_test_counts("1 test passing")
assert counts == [1]
def test_extract_test_counts_empty():
"""Returns empty list when no patterns match."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_test_counts,
)
assert _extract_test_counts("No numbers here.") == []
# ===========================================================================
# _count_test_functions
# ===========================================================================
def test_count_test_functions_basic(tmp_path):
"""Counts def test_ functions across files."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_count_test_functions,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "test_a.py").write_text("def test_one(): pass\ndef test_two(): pass\n", encoding="utf-8")
(tests_dir / "test_b.py").write_text("def test_three(): pass\n", encoding="utf-8")
assert _count_test_functions(tests_dir) == 3
def test_count_test_functions_skips_non_test_files(tmp_path):
"""Only counts from test_*.py files, not conftest or helpers."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_count_test_functions,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
(tests_dir / "test_real.py").write_text("def test_one(): pass\n", encoding="utf-8")
(tests_dir / "conftest.py").write_text("def test_fixture(): pass\n", encoding="utf-8")
(tests_dir / "helpers.py").write_text("def test_helper(): pass\n", encoding="utf-8")
assert _count_test_functions(tests_dir) == 1
def test_count_test_functions_empty_dir(tmp_path):
"""Empty tests dir returns 0."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_count_test_functions,
)
tests_dir = tmp_path / "tests"
tests_dir.mkdir()
assert _count_test_functions(tests_dir) == 0
# ===========================================================================
# 8. check_markdown_links
# ===========================================================================
def test_markdown_links_no_links():
"""README with no relative links passes (skipped)."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
from pathlib import Path
lines = _lines("# Branch\n\nNo links here.\n")
result = check_markdown_links(lines, Path("/tmp"), "fake.py")
assert result["passed"] is True
assert "skipped" in result["message"].lower()
def test_markdown_links_external_only():
"""README with only external links passes (skipped)."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
from pathlib import Path
lines = _lines("[Google](https://google.com)\n[Mail](mailto:a@b.com)\n[Section](#heading)\n")
result = check_markdown_links(lines, Path("/tmp"), "fake.py")
assert result["passed"] is True
assert "skipped" in result["message"].lower()
def test_markdown_links_all_valid(tmp_path):
"""All relative links pointing to existing paths pass."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
(tmp_path / "STATUS.local.md").write_text("# Status\n", encoding="utf-8")
trinity_dir = tmp_path / ".trinity"
trinity_dir.mkdir()
lines = _lines("[Status](STATUS.local.md)\n[Identity](.trinity/)\n")
result = check_markdown_links(lines, tmp_path, "fake.py")
assert result["passed"] is True
assert "2 relative links verified" in result["message"]
def test_markdown_links_dead_link(tmp_path):
"""Dead relative link fails."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
lines = _lines("[Setup](SETUP.md)\n")
result = check_markdown_links(lines, tmp_path, "fake.py")
assert result["passed"] is False
assert "SETUP.md" in result["message"]
def test_markdown_links_mixed_valid_and_dead(tmp_path):
"""Mix of valid and dead links fails, reporting only dead ones."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
(tmp_path / "README.md").write_text("# exists\n", encoding="utf-8")
lines = _lines("[Readme](README.md)\n[Gone](deleted_file.md)\n[Also Gone](tools/)\n")
result = check_markdown_links(lines, tmp_path, "fake.py")
assert result["passed"] is False
assert "deleted_file.md" in result["message"]
assert "tools/" in result["message"]
def test_markdown_links_parent_path(tmp_path):
"""Parent-relative links (../) are resolved correctly."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
parent_file = tmp_path.parent / "parent_readme.md"
parent_file.write_text("# parent\n", encoding="utf-8")
lines = _lines("[Back](../parent_readme.md)\n")
result = check_markdown_links(lines, tmp_path, "fake.py")
assert result["passed"] is True
def test_markdown_links_bypassed():
"""Bypassed standard passes immediately."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
check_markdown_links,
)
from pathlib import Path
bypass_rules = [{"file": "fake.py", "standard": "readme", "reason": "test"}]
lines = _lines("[Dead](nonexistent.md)\n")
result = check_markdown_links(lines, Path("/tmp"), "fake.py", bypass_rules)
assert result["passed"] is True
# ===========================================================================
# _extract_relative_links
# ===========================================================================
def test_extract_relative_links_mixed():
"""Extracts only relative links, skipping external."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_relative_links,
)
content = "[Ext](https://example.com)\n[Local](docs/setup.md)\n[Anchor](#top)\n[File](README.md)\n"
links = _extract_relative_links(content)
assert len(links) == 2
assert ("Local", "docs/setup.md") in links
assert ("File", "README.md") in links
def test_extract_relative_links_empty():
"""No links returns empty list."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_relative_links,
)
assert _extract_relative_links("No links at all.") == []
def test_extract_relative_links_backtick_text():
"""Links with backtick text are extracted correctly."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import (
_extract_relative_links,
)
content = "[`tools/`](tools/)\n"
links = _extract_relative_links(content)
assert len(links) == 1
assert links[0] == ("`tools/`", "tools/")
# ===========================================================================
# Integration: check_module with new checks
# ===========================================================================
def test_check_module_includes_new_checks(tmp_path):
"""check_module result includes test count and link checks."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import check_module
branch_root = tmp_path
apps_dir = branch_root / "apps"
apps_dir.mkdir()
entry = apps_dir / "mybranch.py"
entry.write_text("# entry\n", encoding="utf-8")
readme = branch_root / "README.md"
readme.write_text(
"# MyBranch\n\n"
"## Architecture\n\n```\nmybranch/\n```\n\n"
"## Commands\n\n- `drone @mybranch test`\n\n"
"## Depends On\n\ndrone\n\n"
"*Last Updated: 2099-01-01*\n",
encoding="utf-8",
)
result = check_module(str(entry))
check_names = [c["name"] for c in result["checks"]]
assert "Test count accuracy" in check_names
assert "Markdown link validity" in check_names
assert len(result["checks"]) == 8
def test_check_module_missing_readme_has_8_failures(tmp_path):
"""Missing README produces 8 failure checks (1 exists + 7 dependent)."""
from aipass.seedgo.apps.handlers.aipass_standards.readme_check import check_module
apps_dir = tmp_path / "apps"
apps_dir.mkdir()
entry = apps_dir / "mybranch.py"
entry.write_text("# entry\n", encoding="utf-8")
result = check_module(str(entry))
assert len(result["checks"]) == 8
assert result["score"] == 0