feat(memory): FPLAN-0270 Phase 2 — check_entry validator + drone @memory lint (read-only audit)
Pure check_entry(type,text,limits) validator (chars not bytes, boundary at cap, unknown-type safe) reusable by both gates. New 'drone @memory lint run' scans all branches' .trinity via registry, handles dict+list containers and both key_learning value shapes, sorts worst-first — strictly READ-ONLY (never writes/ trims, honors never_trim_s153). Phase-1 unused_function bypass removed (reader now called). +12 tests (902 total), seedgo 100%. Verified by artifact incl. live lint: 513 over-limit entries across 17 branches (devpulse worst at 71, top offender 5724/600). enforce still false. Part of DPLAN-0205. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIOSAI
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
e47d4f0463
commit
828cc1c8d8
@@ -632,9 +632,19 @@
|
||||
"reason": "Test file — test functions don't require docstrings."
|
||||
},
|
||||
{
|
||||
"file": "apps/handlers/json/entry_limits.py",
|
||||
"standard": "unused_function",
|
||||
"reason": "Phase 1 of FPLAN-0270: load_entry_limits() is deliberately uncalled — Phase 3 wires it into write_memory_file(). Tests cover it."
|
||||
"file": "tests/test_lint.py",
|
||||
"standard": "architecture",
|
||||
"reason": "Test file — lives in tests/ by design, not in 3-layer apps/ structure."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_lint.py",
|
||||
"standard": "documentation",
|
||||
"reason": "Test file — test functions don't require docstrings."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_lint.py",
|
||||
"standard": "meta",
|
||||
"reason": "Test file — META block present at lines 1-7; hook false-positive on test file format."
|
||||
},
|
||||
{
|
||||
"file": "tests/test_entry_limits.py",
|
||||
|
||||
@@ -30,6 +30,9 @@ drone @memory templates push-templates # Push template updates to all branch
|
||||
drone @memory templates diff-templates # Show template differences per branch
|
||||
drone @memory templates template-status # Show template version and push status
|
||||
|
||||
drone @memory lint # Audit .trinity entries for over-limit violations (read-only)
|
||||
drone @memory lint @devpulse # Lint a specific branch
|
||||
|
||||
drone @memory verify FPLAN-XXXX # Check if plan is vectorized in ChromaDB
|
||||
drone @memory watch # Auto-rollover watcher daemon (Ctrl+C to stop)
|
||||
```
|
||||
@@ -42,7 +45,8 @@ drone @memory watch # Auto-rollover watcher daemon (Ctrl+
|
||||
memory/
|
||||
├── apps/
|
||||
│ ├── memory.py # Entry point — auto-discovers modules
|
||||
│ ├── modules/ # 5 modules
|
||||
│ ├── modules/ # 6 modules
|
||||
│ │ ├── lint.py # Entry limit violation scanner (read-only)
|
||||
│ │ ├── rollover.py # Rollover orchestration, status, sync-lines
|
||||
│ │ ├── search.py # Semantic query routing
|
||||
│ │ ├── symbolic.py # Fragmented memory extraction and search
|
||||
@@ -51,7 +55,7 @@ memory/
|
||||
│ └── handlers/ # 14 handler groups
|
||||
│ ├── archive/ # indexer.py
|
||||
│ ├── intake/ # plans_processor.py, pool_processor.py
|
||||
│ ├── json/ # json_handler.py, memory_files.py
|
||||
│ ├── json/ # json_handler.py, memory_files.py, entry_limits.py, lint_handler.py
|
||||
│ ├── learnings/ # manager.py
|
||||
│ ├── monitor/ # detector.py, memory_watcher.py
|
||||
│ ├── rollover/ # extractor.py, orchestrator.py
|
||||
|
||||
@@ -1,26 +1,27 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: entry_limits.py
|
||||
# Description: Entry limits config reader for memory files
|
||||
# Version: 1.0.0
|
||||
# Description: Entry limits config reader and validator for memory files
|
||||
# Version: 1.1.0
|
||||
# Created: 2026-06-13
|
||||
# Modified: 2026-06-13
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Entry Limits Config Reader
|
||||
Entry Limits Config Reader & Validator
|
||||
|
||||
Reads the entry_limits section from memory.config.json and returns
|
||||
the effective limits for a given branch, with per_branch overrides
|
||||
deep-merged over the default entry_types.
|
||||
|
||||
Phase 1 only: reader + safe defaults. No enforcement, no validation,
|
||||
no write-path integration.
|
||||
Provides ``check_entry()`` — a pure validator that checks whether a
|
||||
single entry text exceeds its character cap.
|
||||
|
||||
Usage:
|
||||
from aipass.memory.apps.handlers.json.entry_limits import load_entry_limits
|
||||
from aipass.memory.apps.handlers.json.entry_limits import load_entry_limits, check_entry
|
||||
|
||||
limits = load_entry_limits("devpulse")
|
||||
# => {"enabled": True, "enforce": False, "entry_types": {...}}
|
||||
verdict = check_entry("key_learnings", some_text, limits)
|
||||
# => {"ok": True/False, "length": int, "cap": int, "over_by": int, "entry_type": str}
|
||||
"""
|
||||
|
||||
import copy
|
||||
@@ -188,3 +189,57 @@ def load_entry_limits(branch: str) -> dict[str, Any]:
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Phase 2: pure entry validator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def check_entry(entry_type: str, text: str, limits: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Check whether *text* exceeds the character cap for *entry_type*.
|
||||
|
||||
This is a **pure function** — no I/O, no file reads, no side effects
|
||||
(except a debug log when *entry_type* is unknown).
|
||||
|
||||
Args:
|
||||
entry_type: Name of the entry type (e.g. ``"key_learnings"``).
|
||||
text: The entry text to measure.
|
||||
limits: The dict returned by :func:`load_entry_limits`.
|
||||
|
||||
Returns:
|
||||
Verdict dict::
|
||||
|
||||
{
|
||||
"ok": bool, # True when within cap (length <= cap)
|
||||
"length": int, # len(text) — characters, not bytes
|
||||
"cap": int, # max_chars for this type (0 if unknown)
|
||||
"over_by": int, # max(0, length - cap)
|
||||
"entry_type": str, # echo back the entry_type
|
||||
}
|
||||
"""
|
||||
entry_types = limits.get("entry_types", {})
|
||||
type_def = entry_types.get(entry_type)
|
||||
|
||||
length = len(text)
|
||||
|
||||
if type_def is None:
|
||||
logger.info(f"[entry_limits] Unknown entry_type '{entry_type}' — no cap applied")
|
||||
return {
|
||||
"ok": True,
|
||||
"length": length,
|
||||
"cap": 0,
|
||||
"over_by": 0,
|
||||
"entry_type": entry_type,
|
||||
}
|
||||
|
||||
cap = type_def.get("max_chars", 0)
|
||||
over_by = max(0, length - cap)
|
||||
|
||||
return {
|
||||
"ok": length <= cap,
|
||||
"length": length,
|
||||
"cap": cap,
|
||||
"over_by": over_by,
|
||||
"entry_type": entry_type,
|
||||
}
|
||||
|
||||
@@ -0,0 +1,220 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: lint_handler.py
|
||||
# Description: Read-only lint handler for .trinity entry limit violations
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-06-13
|
||||
# Modified: 2026-06-13
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Lint Handler — Entry Limit Violation Scanner
|
||||
|
||||
Scans .trinity memory files across branches and reports entries that
|
||||
exceed their configured character caps. Strictly **read-only** — never
|
||||
writes, modifies, truncates, or deletes any file.
|
||||
|
||||
Called by the ``lint`` module (thin CLI layer).
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.memory.apps.handlers.json import json_handler
|
||||
from aipass.memory.apps.handlers.json.entry_limits import (
|
||||
check_entry,
|
||||
load_entry_limits,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _measure_dict_container(
|
||||
data: dict[str, Any],
|
||||
field: str,
|
||||
) -> list[tuple[str, str]]:
|
||||
"""Extract (key, text) pairs from a dict-style container.
|
||||
|
||||
Each value may be:
|
||||
- a plain string (the entry itself), or
|
||||
- a dict containing *field* (the entry is ``value[field]``).
|
||||
|
||||
Returns a list of ``(key, text)`` tuples for measurable entries.
|
||||
"""
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for key, value in data.items():
|
||||
if isinstance(value, str):
|
||||
pairs.append((key, value))
|
||||
elif isinstance(value, dict):
|
||||
if field in value:
|
||||
pairs.append((key, value[field]))
|
||||
return pairs
|
||||
|
||||
|
||||
def _measure_list_container(
|
||||
data: list[Any],
|
||||
field: str,
|
||||
) -> list[tuple[str, str]]:
|
||||
"""Extract (index-label, text) pairs from a list-style container.
|
||||
|
||||
Each item is expected to be a dict containing *field*. Items that
|
||||
are not dicts or lack the field are silently skipped.
|
||||
|
||||
Returns a list of ``("[idx]", text)`` tuples.
|
||||
"""
|
||||
pairs: list[tuple[str, str]] = []
|
||||
for idx, item in enumerate(data):
|
||||
if isinstance(item, dict) and field in item:
|
||||
pairs.append((f"[{idx}]", item[field]))
|
||||
return pairs
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Core lint logic
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _lint_branch(
|
||||
branch_name: str,
|
||||
branch_path: str,
|
||||
limits: dict[str, Any],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Lint a single branch and return a list of violation dicts.
|
||||
|
||||
Each violation dict has keys:
|
||||
branch, file, container, key, length, cap, over_by, entry_type
|
||||
"""
|
||||
violations: list[dict[str, Any]] = []
|
||||
trinity_dir = Path(branch_path) / ".trinity"
|
||||
|
||||
if not trinity_dir.is_dir():
|
||||
logger.info(f"[lint] Branch '{branch_name}' has no .trinity directory, skipping")
|
||||
return violations
|
||||
|
||||
entry_types = limits.get("entry_types", {})
|
||||
|
||||
for type_name, type_def in entry_types.items():
|
||||
file_name = type_def.get("file", "")
|
||||
container = type_def.get("container", "")
|
||||
kind = type_def.get("kind", "")
|
||||
field = type_def.get("field", "")
|
||||
|
||||
file_path = trinity_dir / file_name
|
||||
if not file_path.is_file():
|
||||
logger.info(f"[lint] {branch_name}: missing {file_name}, skipping {type_name}")
|
||||
continue
|
||||
|
||||
try:
|
||||
raw = file_path.read_text(encoding="utf-8")
|
||||
data = json.loads(raw)
|
||||
except (json.JSONDecodeError, OSError) as exc:
|
||||
logger.warning(f"[lint] {branch_name}: failed to read {file_name}: {exc}")
|
||||
continue
|
||||
|
||||
container_data = data.get(container)
|
||||
if container_data is None:
|
||||
continue
|
||||
|
||||
# Build (key, text) pairs depending on kind
|
||||
if kind == "dict" and isinstance(container_data, dict):
|
||||
pairs = _measure_dict_container(container_data, field)
|
||||
elif kind == "list" and isinstance(container_data, list):
|
||||
pairs = _measure_list_container(container_data, field)
|
||||
else:
|
||||
continue
|
||||
|
||||
for key, text in pairs:
|
||||
verdict = check_entry(type_name, text, limits)
|
||||
if not verdict["ok"]:
|
||||
violations.append(
|
||||
{
|
||||
"branch": branch_name,
|
||||
"file": file_name,
|
||||
"container": container,
|
||||
"key": key,
|
||||
"length": verdict["length"],
|
||||
"cap": verdict["cap"],
|
||||
"over_by": verdict["over_by"],
|
||||
"entry_type": type_name,
|
||||
}
|
||||
)
|
||||
|
||||
return violations
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def run_lint(
|
||||
branches: list[dict[str, Any]],
|
||||
branch_filter: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Scan branches for entry-limit violations.
|
||||
|
||||
This function is **read-only** — it never writes, modifies, truncates,
|
||||
or deletes any file.
|
||||
|
||||
Args:
|
||||
branches: List of branch dicts (``{"name": ..., "path": ...}``),
|
||||
typically from ``_read_registry()`` in the module layer.
|
||||
branch_filter: If provided, only lint this branch (case-insensitive).
|
||||
|
||||
Returns:
|
||||
Result dict::
|
||||
|
||||
{
|
||||
"success": True,
|
||||
"violations": [...], # sorted worst-first (highest over_by)
|
||||
"total_violations": int,
|
||||
"branches_scanned": int,
|
||||
"branches_skipped": int,
|
||||
}
|
||||
"""
|
||||
all_violations: list[dict[str, Any]] = []
|
||||
branches_scanned = 0
|
||||
branches_skipped = 0
|
||||
|
||||
for branch in branches:
|
||||
name = branch.get("name", "unknown")
|
||||
path = branch.get("path", "")
|
||||
|
||||
# Apply branch filter (case-insensitive)
|
||||
if branch_filter and name.lower() != branch_filter.lower():
|
||||
continue
|
||||
|
||||
limits = load_entry_limits(name)
|
||||
|
||||
if not limits.get("enabled", True):
|
||||
branches_skipped += 1
|
||||
continue
|
||||
|
||||
branch_violations = _lint_branch(name, path, limits)
|
||||
all_violations.extend(branch_violations)
|
||||
branches_scanned += 1
|
||||
|
||||
# Sort worst-first (highest over_by)
|
||||
all_violations.sort(key=lambda v: v["over_by"], reverse=True)
|
||||
|
||||
json_handler.log_operation(
|
||||
"lint",
|
||||
{
|
||||
"total_violations": len(all_violations),
|
||||
"branches_scanned": branches_scanned,
|
||||
"branch_filter": branch_filter,
|
||||
},
|
||||
module_name="lint",
|
||||
)
|
||||
|
||||
return {
|
||||
"success": True,
|
||||
"violations": all_violations,
|
||||
"total_violations": len(all_violations),
|
||||
"branches_scanned": branches_scanned,
|
||||
"branches_skipped": branches_skipped,
|
||||
}
|
||||
@@ -82,7 +82,8 @@ def print_help():
|
||||
console.print()
|
||||
console.print(
|
||||
Panel.fit(
|
||||
"[bold cyan]Memory - Central Memory Archive System[/bold cyan]\n[dim]Vector search, memory rollover, and fragmented memory for AIPass[/dim]",
|
||||
"[bold cyan]Memory - Central Memory Archive System[/bold cyan]\n"
|
||||
"[dim]Vector search, memory rollover, and fragmented memory for AIPass[/dim]",
|
||||
border_style="cyan",
|
||||
box=box.ROUNDED,
|
||||
)
|
||||
@@ -120,6 +121,7 @@ def print_help():
|
||||
table.add_row("pool process", "Process pool files + check/run rollover")
|
||||
table.add_row("pool status", "Show pool file count, config, vector stats")
|
||||
table.add_row("verify <plan_label>", "Check if a plan is vectorized in ChromaDB")
|
||||
table.add_row("lint [@branch]", "Audit .trinity entries for over-limit violations (read-only)")
|
||||
table.add_row("watch", "Start memory watcher (auto-rollover on changes)")
|
||||
|
||||
console.print(table)
|
||||
@@ -166,7 +168,8 @@ def print_help():
|
||||
console.print()
|
||||
|
||||
console.print(
|
||||
"Commands: search, rollover [run|status|check|sync-lines], pool [process|status], symbolic, templates, verify, watch"
|
||||
"Commands: search, rollover [run|status|check|sync-lines], lint,"
|
||||
" pool [process|status], symbolic, templates, verify, watch"
|
||||
)
|
||||
console.print()
|
||||
|
||||
|
||||
@@ -0,0 +1,237 @@
|
||||
# =================== AIPass ====================
|
||||
# Name: lint.py
|
||||
# Description: Lint module — CLI routing for entry limit auditing
|
||||
# Version: 1.0.0
|
||||
# Created: 2026-06-13
|
||||
# Modified: 2026-06-13
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Lint Module — Entry Limit Violation Scanner
|
||||
|
||||
Thin CLI routing layer that discovers branches via the registry,
|
||||
delegates scanning to the lint handler, and formats results for
|
||||
the console.
|
||||
|
||||
Strictly **read-only** — never writes, modifies, truncates, or
|
||||
deletes any file.
|
||||
|
||||
Usage:
|
||||
drone @memory lint # Scan all branches
|
||||
drone @memory lint @devpulse # Scan one branch
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
|
||||
from aipass.prax import logger
|
||||
from aipass.cli.apps.modules import console, error, warning
|
||||
from aipass.memory.apps.handlers.json import json_handler
|
||||
|
||||
# Handler import (same package family — json handlers)
|
||||
from aipass.memory.apps.handlers.json.lint_handler import run_lint
|
||||
|
||||
# Cross-handler access for branch discovery (module layer bridges handlers)
|
||||
from aipass.memory.apps.handlers.monitor.detector import _read_registry
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# COMMAND HANDLER
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def handle_command(command: str, args: list[str]) -> bool:
|
||||
"""Handle lint commands with seedgo-compliant introspection.
|
||||
|
||||
Routing:
|
||||
lint (no args) -> print_introspection()
|
||||
lint --help / -h / help -> print_help()
|
||||
lint @branch -> scan one branch
|
||||
lint run -> scan all branches
|
||||
lint run @branch -> scan one branch
|
||||
|
||||
Args:
|
||||
command: Command name.
|
||||
args: Additional arguments.
|
||||
|
||||
Returns:
|
||||
True if command handled, False otherwise.
|
||||
"""
|
||||
if command != "lint":
|
||||
return False
|
||||
|
||||
# No args -> introspection (seedgo standard)
|
||||
if not args:
|
||||
print_introspection()
|
||||
return True
|
||||
|
||||
# Help
|
||||
if args[0] in ("--help", "-h", "help"):
|
||||
print_help()
|
||||
return True
|
||||
|
||||
# Parse optional @branch filter
|
||||
branch_filter = _extract_branch(args)
|
||||
|
||||
# "run" subcommand is accepted but optional — lint always runs
|
||||
filtered_args = [a for a in args if a != "run" and not a.startswith("@")]
|
||||
|
||||
if filtered_args:
|
||||
error(
|
||||
f"Unknown lint argument: {filtered_args[0]}",
|
||||
suggestion="Run 'drone @memory lint help' for usage",
|
||||
)
|
||||
return True
|
||||
|
||||
_execute_lint(branch_filter)
|
||||
return True
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# ARGUMENT HELPERS
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _extract_branch(args: list[str]) -> str | None:
|
||||
"""Extract @branch from args, return branch name or None."""
|
||||
for arg in args:
|
||||
if arg.startswith("@"):
|
||||
return arg[1:]
|
||||
return None
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# LINT EXECUTION
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _execute_lint(branch_filter: str | None = None) -> None:
|
||||
"""Run the lint scan and display results.
|
||||
|
||||
Args:
|
||||
branch_filter: If provided, only lint this branch.
|
||||
"""
|
||||
# Branch discovery happens in the module layer (bridges handlers)
|
||||
try:
|
||||
branches = _read_registry()
|
||||
except Exception as exc:
|
||||
logger.warning(f"[lint] Failed to read registry: {exc}")
|
||||
error(f"Failed to read registry: {exc}")
|
||||
return
|
||||
|
||||
if not branches:
|
||||
warning("No branches found in registry")
|
||||
return
|
||||
|
||||
result = run_lint(branches, branch_filter=branch_filter)
|
||||
|
||||
if not result.get("success"):
|
||||
error(result.get("error", "Unknown lint error"))
|
||||
return
|
||||
|
||||
_display_results(result, branch_filter)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# DISPLAY
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def _display_results(result: dict[str, Any], branch_filter: str | None) -> None:
|
||||
"""Format and display lint results via Rich console.
|
||||
|
||||
Args:
|
||||
result: Result dict from ``run_lint``.
|
||||
branch_filter: The branch filter used (for display context).
|
||||
"""
|
||||
violations = result.get("violations", [])
|
||||
scanned = result.get("branches_scanned", 0)
|
||||
skipped = result.get("branches_skipped", 0)
|
||||
total = result.get("total_violations", 0)
|
||||
|
||||
console.print()
|
||||
|
||||
if not violations:
|
||||
scope = f"@{branch_filter}" if branch_filter else "all branches"
|
||||
console.print(f"[green]No violations found[/green] across {scope} ({scanned} scanned)")
|
||||
console.print()
|
||||
return
|
||||
|
||||
# Per-violation detail (sorted worst-first by handler)
|
||||
console.print(f"[bold red]{total} violation(s) found[/bold red]")
|
||||
console.print()
|
||||
|
||||
current_branch: str | None = None
|
||||
branch_count = 0
|
||||
|
||||
for v in violations:
|
||||
branch = v["branch"]
|
||||
if branch != current_branch:
|
||||
if current_branch is not None:
|
||||
console.print()
|
||||
console.print(f" [bold cyan]{branch}[/bold cyan]")
|
||||
current_branch = branch
|
||||
branch_count = 0
|
||||
|
||||
branch_count += 1
|
||||
console.print(
|
||||
f" [red]![/red] {v['file']}:{v['container']}/{v['key']} "
|
||||
f"[dim]({v['entry_type']})[/dim] "
|
||||
f"{v['length']}/{v['cap']} chars "
|
||||
f"[red]+{v['over_by']} over[/red]"
|
||||
)
|
||||
|
||||
console.print()
|
||||
console.print(f"[dim]Scanned {scanned} branch(es), skipped {skipped}[/dim]")
|
||||
console.print()
|
||||
|
||||
json_handler.log_operation(
|
||||
"lint_display",
|
||||
{"total_violations": total, "branches_scanned": scanned},
|
||||
module_name="lint",
|
||||
)
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# INTROSPECTION
|
||||
# =============================================================================
|
||||
|
||||
|
||||
def print_introspection() -> None:
|
||||
"""Display module introspection (seedgo standard).
|
||||
|
||||
Called when ``lint`` is invoked with no arguments.
|
||||
"""
|
||||
console.print()
|
||||
console.print("[bold cyan]lint Module[/bold cyan]")
|
||||
console.print("Audits .trinity entries for over-limit character violations (read-only)")
|
||||
console.print()
|
||||
|
||||
console.print("[yellow]Connected Handlers:[/yellow]")
|
||||
console.print(" [cyan]handlers/json/[/cyan] [dim]lint_handler.py, entry_limits.py[/dim]")
|
||||
console.print()
|
||||
|
||||
console.print("[yellow]Next:[/yellow]")
|
||||
console.print(" [green]drone @memory lint run[/green] [dim]# Scan all branches[/dim]")
|
||||
console.print(" [green]drone @memory lint @devpulse[/green] [dim]# Scan one branch[/dim]")
|
||||
console.print(" [green]drone @memory lint help[/green] [dim]# Full usage guide[/dim]")
|
||||
console.print()
|
||||
|
||||
|
||||
def print_help() -> None:
|
||||
"""Display lint module help."""
|
||||
console.print()
|
||||
console.print("[bold cyan]Lint Module - Entry Limit Violation Scanner[/bold cyan]")
|
||||
console.print()
|
||||
console.print("[bold]USAGE:[/bold]")
|
||||
console.print(" drone @memory lint Scan all branches")
|
||||
console.print(" drone @memory lint @<branch> Scan a specific branch")
|
||||
console.print(" drone @memory lint run Scan all branches (explicit)")
|
||||
console.print()
|
||||
console.print("[bold]WHAT IT DOES:[/bold]")
|
||||
console.print(" Reads .trinity/local.json and .trinity/observations.json for every")
|
||||
console.print(" registered branch. Checks each entry against configured character")
|
||||
console.print(" caps from memory.config.json. Reports violations sorted worst-first.")
|
||||
console.print()
|
||||
console.print("[bold]NOTE:[/bold]")
|
||||
console.print(" This command is strictly [green]read-only[/green]. It never modifies any file.")
|
||||
console.print()
|
||||
@@ -0,0 +1,409 @@
|
||||
# ===================AIPASS====================
|
||||
# META DATA HEADER
|
||||
# Name: tests/test_lint.py
|
||||
# Date: 2026-06-13
|
||||
# Version: 1.0.0
|
||||
# Category: memory/tests
|
||||
# =============================================
|
||||
|
||||
"""
|
||||
Tests for Phase 2 of FPLAN-0270: check_entry validator + lint handler.
|
||||
|
||||
Covers:
|
||||
- check_entry boundary checks (at-cap, cap+1, larger over)
|
||||
- Character-not-byte counting (em-dash, tree glyphs)
|
||||
- Unknown entry_type handling
|
||||
- Dict container measurement (plain string + dict-with-field)
|
||||
- List container measurement + missing field skip
|
||||
- Lint handler finds violations with correct counts
|
||||
- Lint handler is read-only (files unchanged after scan)
|
||||
"""
|
||||
|
||||
import importlib
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers: fresh-import modules under test with mocks already in place
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _fresh_lint_modules():
|
||||
"""Drop cached modules so each test gets a fresh import."""
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json", None)
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json.json_handler", None)
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json.entry_limits", None)
|
||||
sys.modules.pop("aipass.memory.apps.handlers.json.lint_handler", None)
|
||||
sys.modules.pop("aipass.memory.apps.modules.lint", None)
|
||||
yield
|
||||
|
||||
|
||||
def _get_entry_limits():
|
||||
"""Import and return the entry_limits module."""
|
||||
return importlib.import_module("aipass.memory.apps.handlers.json.entry_limits")
|
||||
|
||||
|
||||
def _get_lint_handler():
|
||||
"""Import and return the lint_handler module."""
|
||||
return importlib.import_module("aipass.memory.apps.handlers.json.lint_handler")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_limits(entry_types: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
"""Build a limits dict matching the shape returned by load_entry_limits."""
|
||||
if entry_types is None:
|
||||
entry_types = {
|
||||
"key_learnings": {
|
||||
"file": "local.json",
|
||||
"container": "key_learnings",
|
||||
"kind": "dict",
|
||||
"field": "value",
|
||||
"max_chars": 10,
|
||||
},
|
||||
}
|
||||
return {"enabled": True, "enforce": False, "entry_types": entry_types}
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 1. check_entry tests
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestCheckEntryAtCap:
|
||||
"""length == cap is OK (not over)."""
|
||||
|
||||
def test_at_cap_is_ok(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
limits = _make_limits()
|
||||
text = "a" * 10 # exactly at cap
|
||||
|
||||
result = mod.check_entry("key_learnings", text, limits)
|
||||
|
||||
assert result["ok"] is True
|
||||
assert result["length"] == 10
|
||||
assert result["cap"] == 10
|
||||
assert result["over_by"] == 0
|
||||
assert result["entry_type"] == "key_learnings"
|
||||
|
||||
|
||||
class TestCheckEntryCapPlusOne:
|
||||
"""length == cap+1 is OVER."""
|
||||
|
||||
def test_cap_plus_one_is_over(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
limits = _make_limits()
|
||||
text = "a" * 11 # one over cap
|
||||
|
||||
result = mod.check_entry("key_learnings", text, limits)
|
||||
|
||||
assert result["ok"] is False
|
||||
assert result["length"] == 11
|
||||
assert result["cap"] == 10
|
||||
assert result["over_by"] == 1
|
||||
|
||||
|
||||
class TestCheckEntryLargerOver:
|
||||
"""over_by calculation correct for strings well over cap."""
|
||||
|
||||
def test_over_by_large(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
limits = _make_limits()
|
||||
text = "a" * 25 # 15 over cap of 10
|
||||
|
||||
result = mod.check_entry("key_learnings", text, limits)
|
||||
|
||||
assert result["ok"] is False
|
||||
assert result["length"] == 25
|
||||
assert result["over_by"] == 15
|
||||
|
||||
|
||||
class TestCheckEntryCharNotByte:
|
||||
"""Em-dash is 3 bytes UTF-8 but 1 character -- count chars not bytes."""
|
||||
|
||||
def test_em_dash_counts_as_one_char(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
# "a—b" is 3 characters, not 5 bytes
|
||||
text = "a—b"
|
||||
assert len(text) == 3
|
||||
assert len(text.encode("utf-8")) == 5 # prove multi-byte
|
||||
|
||||
limits = _make_limits()
|
||||
result = mod.check_entry("key_learnings", text, limits)
|
||||
|
||||
assert result["length"] == 3 # chars, not bytes
|
||||
assert result["ok"] is True
|
||||
|
||||
def test_tree_glyph_counts_as_one_char(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
# tree glyph is multi-byte UTF-8 but one character
|
||||
text = "a└b"
|
||||
assert len(text) == 3
|
||||
assert len(text.encode("utf-8")) == 5
|
||||
|
||||
limits = _make_limits()
|
||||
result = mod.check_entry("key_learnings", text, limits)
|
||||
|
||||
assert result["length"] == 3
|
||||
|
||||
|
||||
class TestCheckEntryUnknownType:
|
||||
"""Unknown entry_type returns ok=True, cap=0."""
|
||||
|
||||
def test_unknown_type_always_ok(self) -> None:
|
||||
mod = _get_entry_limits()
|
||||
limits = _make_limits()
|
||||
|
||||
result = mod.check_entry("nonexistent_type", "any text", limits)
|
||||
|
||||
assert result["ok"] is True
|
||||
assert result["cap"] == 0
|
||||
assert result["over_by"] == 0
|
||||
assert result["entry_type"] == "nonexistent_type"
|
||||
assert result["length"] == len("any text")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 2. Container handling tests
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestDictContainerStringValue:
|
||||
"""Dict container where value is a plain string (key_learnings style)."""
|
||||
|
||||
def test_dict_string_value_measured(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
# Build a branch with a dict container whose values are plain strings
|
||||
trinity = tmp_path / "branch" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
local_data = {
|
||||
"key_learnings": {
|
||||
"learn1": "short", # 5 chars, under cap of 10
|
||||
"learn2": "this is way too long for the cap", # over
|
||||
},
|
||||
}
|
||||
(trinity / "local.json").write_text(json.dumps(local_data), encoding="utf-8")
|
||||
|
||||
limits = _make_limits()
|
||||
violations = handler._lint_branch("test", str(tmp_path / "branch"), limits)
|
||||
|
||||
assert len(violations) == 1
|
||||
assert violations[0]["key"] == "learn2"
|
||||
assert violations[0]["container"] == "key_learnings"
|
||||
|
||||
|
||||
class TestDictContainerDictValue:
|
||||
"""Dict container where value is a dict with a field key."""
|
||||
|
||||
def test_dict_with_field_measured(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
trinity = tmp_path / "branch" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
local_data = {
|
||||
"key_learnings": {
|
||||
"learn1": {"value": "ok", "meta": "x"}, # 2 chars
|
||||
"learn2": {"value": "this exceeds the limit!!", "meta": "y"}, # over
|
||||
},
|
||||
}
|
||||
(trinity / "local.json").write_text(json.dumps(local_data), encoding="utf-8")
|
||||
|
||||
limits = _make_limits(
|
||||
{
|
||||
"key_learnings": {
|
||||
"file": "local.json",
|
||||
"container": "key_learnings",
|
||||
"kind": "dict",
|
||||
"field": "value",
|
||||
"max_chars": 10,
|
||||
},
|
||||
}
|
||||
)
|
||||
violations = handler._lint_branch("test", str(tmp_path / "branch"), limits)
|
||||
|
||||
assert len(violations) == 1
|
||||
assert violations[0]["key"] == "learn2"
|
||||
|
||||
|
||||
class TestListContainer:
|
||||
"""List container (sessions/observations style)."""
|
||||
|
||||
def test_list_items_measured(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
trinity = tmp_path / "branch" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
obs_data = {
|
||||
"observations": [
|
||||
{"note": "short"}, # 5 chars
|
||||
{"note": "this observation is way too long for the cap"}, # over
|
||||
],
|
||||
}
|
||||
(trinity / "observations.json").write_text(json.dumps(obs_data), encoding="utf-8")
|
||||
|
||||
limits = _make_limits(
|
||||
{
|
||||
"obs": {
|
||||
"file": "observations.json",
|
||||
"container": "observations",
|
||||
"kind": "list",
|
||||
"field": "note",
|
||||
"max_chars": 10,
|
||||
},
|
||||
}
|
||||
)
|
||||
violations = handler._lint_branch("test", str(tmp_path / "branch"), limits)
|
||||
|
||||
assert len(violations) == 1
|
||||
assert violations[0]["key"] == "[1]"
|
||||
assert violations[0]["entry_type"] == "obs"
|
||||
|
||||
|
||||
class TestListContainerMissingField:
|
||||
"""Missing field in a list item is skipped, no crash."""
|
||||
|
||||
def test_missing_field_skipped(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
trinity = tmp_path / "branch" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
obs_data = {
|
||||
"observations": [
|
||||
{"note": "short"}, # has field
|
||||
{"other_key": "no note here"}, # missing field
|
||||
{"note": "also short"}, # has field
|
||||
],
|
||||
}
|
||||
(trinity / "observations.json").write_text(json.dumps(obs_data), encoding="utf-8")
|
||||
|
||||
limits = _make_limits(
|
||||
{
|
||||
"obs": {
|
||||
"file": "observations.json",
|
||||
"container": "observations",
|
||||
"kind": "list",
|
||||
"field": "note",
|
||||
"max_chars": 100,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
# Should not crash, and no violations (all within cap)
|
||||
violations = handler._lint_branch("test", str(tmp_path / "branch"), limits)
|
||||
assert len(violations) == 0
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# 3. Lint handler integration tests
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestLintHandlerFindsViolations:
|
||||
"""Lint handler finds planted violations with correct counts."""
|
||||
|
||||
def test_finds_violations(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
# Create a branch with planted over-limit entries
|
||||
trinity = tmp_path / "branch_a" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
|
||||
local_data = {
|
||||
"key_learnings": {
|
||||
"ok_entry": "fine",
|
||||
"bad_entry": "x" * 15, # 15 chars, cap 10 -> over by 5
|
||||
},
|
||||
}
|
||||
(trinity / "local.json").write_text(json.dumps(local_data), encoding="utf-8")
|
||||
|
||||
branches = [{"name": "branch_a", "path": str(tmp_path / "branch_a")}]
|
||||
limits = _make_limits()
|
||||
|
||||
# Monkeypatch load_entry_limits to return our test limits
|
||||
with patch.object(handler, "load_entry_limits", return_value=limits):
|
||||
result = handler.run_lint(branches)
|
||||
|
||||
assert result["success"] is True
|
||||
assert result["total_violations"] == 1
|
||||
assert result["branches_scanned"] == 1
|
||||
|
||||
v = result["violations"][0]
|
||||
assert v["branch"] == "branch_a"
|
||||
assert v["key"] == "bad_entry"
|
||||
assert v["over_by"] == 5
|
||||
assert v["length"] == 15
|
||||
assert v["cap"] == 10
|
||||
|
||||
|
||||
class TestLintHandlerReadOnly:
|
||||
"""Lint handler must be strictly read-only -- files unchanged after scan."""
|
||||
|
||||
def test_files_unchanged_after_lint(self, tmp_path: Path) -> None:
|
||||
handler = _get_lint_handler()
|
||||
|
||||
# Create branch with violations
|
||||
trinity = tmp_path / "branch_b" / ".trinity"
|
||||
trinity.mkdir(parents=True)
|
||||
|
||||
local_data = {
|
||||
"key_learnings": {
|
||||
"big": "x" * 50,
|
||||
},
|
||||
}
|
||||
local_path = trinity / "local.json"
|
||||
local_content = json.dumps(local_data, indent=2)
|
||||
local_path.write_text(local_content, encoding="utf-8")
|
||||
|
||||
obs_data = {
|
||||
"observations": [
|
||||
{"note": "y" * 50},
|
||||
],
|
||||
}
|
||||
obs_path = trinity / "observations.json"
|
||||
obs_content = json.dumps(obs_data, indent=2)
|
||||
obs_path.write_text(obs_content, encoding="utf-8")
|
||||
|
||||
# Read content before lint
|
||||
local_before = local_path.read_text(encoding="utf-8")
|
||||
obs_before = obs_path.read_text(encoding="utf-8")
|
||||
|
||||
branches = [{"name": "branch_b", "path": str(tmp_path / "branch_b")}]
|
||||
limits = _make_limits(
|
||||
{
|
||||
"key_learnings": {
|
||||
"file": "local.json",
|
||||
"container": "key_learnings",
|
||||
"kind": "dict",
|
||||
"field": "value",
|
||||
"max_chars": 10,
|
||||
},
|
||||
"observations": {
|
||||
"file": "observations.json",
|
||||
"container": "observations",
|
||||
"kind": "list",
|
||||
"field": "note",
|
||||
"max_chars": 10,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
with patch.object(handler, "load_entry_limits", return_value=limits):
|
||||
handler.run_lint(branches)
|
||||
|
||||
# Assert files are UNCHANGED
|
||||
local_after = local_path.read_text(encoding="utf-8")
|
||||
obs_after = obs_path.read_text(encoding="utf-8")
|
||||
|
||||
assert local_before == local_after, "local.json was modified by lint!"
|
||||
assert obs_before == obs_after, "observations.json was modified by lint!"
|
||||
Reference in New Issue
Block a user