diff --git a/src/aipass/memory/.seedgo/bypass.json b/src/aipass/memory/.seedgo/bypass.json index 64f90f87..1dc3a82a 100644 --- a/src/aipass/memory/.seedgo/bypass.json +++ b/src/aipass/memory/.seedgo/bypass.json @@ -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", diff --git a/src/aipass/memory/README.md b/src/aipass/memory/README.md index 5991065c..72a261dd 100644 --- a/src/aipass/memory/README.md +++ b/src/aipass/memory/README.md @@ -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 diff --git a/src/aipass/memory/apps/handlers/json/entry_limits.py b/src/aipass/memory/apps/handlers/json/entry_limits.py index 088b86f5..d5a21dd5 100644 --- a/src/aipass/memory/apps/handlers/json/entry_limits.py +++ b/src/aipass/memory/apps/handlers/json/entry_limits.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, + } diff --git a/src/aipass/memory/apps/handlers/json/lint_handler.py b/src/aipass/memory/apps/handlers/json/lint_handler.py new file mode 100644 index 00000000..14f62f62 --- /dev/null +++ b/src/aipass/memory/apps/handlers/json/lint_handler.py @@ -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, + } diff --git a/src/aipass/memory/apps/memory.py b/src/aipass/memory/apps/memory.py index 2d67f451..7c37a4c0 100755 --- a/src/aipass/memory/apps/memory.py +++ b/src/aipass/memory/apps/memory.py @@ -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 ", "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() diff --git a/src/aipass/memory/apps/modules/lint.py b/src/aipass/memory/apps/modules/lint.py new file mode 100644 index 00000000..f057476b --- /dev/null +++ b/src/aipass/memory/apps/modules/lint.py @@ -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 @ 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() diff --git a/src/aipass/memory/tests/test_lint.py b/src/aipass/memory/tests/test_lint.py new file mode 100644 index 00000000..28055a5d --- /dev/null +++ b/src/aipass/memory/tests/test_lint.py @@ -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!"