Nav Loop

#!/usr/bin/env python3
"""
Dual-condition exit gate for Navigator loop mode.

Evaluates whether loop should exit based on:
1. Heuristics: At least N completion indicators met
2. Explicit signal: EXIT_SIGNAL must be true

Usage:
    python3 exit_gate.py \
        --indicators '{"code_committed": true, "tests_passing": true}' \
        --exit-signal \
        --min-heuristics 2

Output:
    JSON with exit decision and reasoning
"""

import argparse
import json
import sys
from typing import Tuple


def count_indicators(indicators: dict) -> Tuple[int, int]:
    """Count met indicators vs total."""
    if not indicators:
        return 0, 5

    met = sum(1 for v in indicators.values() if v)
    total = len(indicators)
    return met, total


def evaluate_exit(
    indicators: dict,
    exit_signal: bool,
    min_heuristics: int = 2,
    require_explicit: bool = True
) -> dict:
    """
    Evaluate dual-condition exit gate.

    Returns dict with:
        - should_exit: bool
        - reason: str
        - heuristics_met: int
        - heuristics_total: int
        - exit_signal: bool
        - blocked_reason: str or None
    """
    met, total = count_indicators(indicators)
    heuristics_satisfied = met >= min_heuristics

    result = {
        "heuristics_met": met,
        "heuristics_total": total,
        "heuristics_satisfied": heuristics_satisfied,
        "exit_signal": exit_signal,
        "min_required": min_heuristics,
        "require_explicit": require_explicit
    }

    # Dual-condition evaluation
    if heuristics_satisfied and exit_signal:
        result["should_exit"] = True
        result["reason"] = f"EXIT: {met}/{total} heuristics + explicit signal"
        result["blocked_reason"] = None

    elif exit_signal and not heuristics_satisfied:
        result["should_exit"] = False
        result["reason"] = f"BLOCKED: Exit signal but only {met}/{total} heuristics"
        result["blocked_reason"] = "Insufficient completion indicators"

    elif heuristics_satisfied and not exit_signal:
        if require_explicit:
            result["should_exit"] = False
            result["reason"] = f"CONTINUE: {met}/{total} heuristics, awaiting EXIT_SIGNAL"
            result["blocked_reason"] = "Awaiting explicit completion signal"
        else:
            # Legacy mode: exit on heuristics alone
            result["should_exit"] = True
            result["reason"] = f"EXIT: {met}/{total} heuristics (explicit signal not required)"
            result["blocked_reason"] = None

    else:
        result["should_exit"] = False
        result["reason"] = f"CONTINUE: {met}/{total} heuristics, no exit signal"
        result["blocked_reason"] = "More work needed"

    return result


def main():
    parser = argparse.ArgumentParser(
        description="Evaluate dual-condition exit gate"
    )
    parser.add_argument("--indicators", default="{}",
                        help="JSON object of indicator states")
    parser.add_argument("--exit-signal", action="store_true",
                        help="Whether EXIT_SIGNAL was explicitly set")
    parser.add_argument("--min-heuristics", type=int, default=2,
                        help="Minimum indicators required (default: 2)")
    parser.add_argument("--no-require-explicit", action="store_true",
                        help="Don't require explicit EXIT_SIGNAL")
    parser.add_argument("--output", choices=["json", "text"], default="json",
                        help="Output format")

    args = parser.parse_args()

    try:
        indicators = json.loads(args.indicators)
    except json.JSONDecodeError:
        print("Error: Invalid JSON for indicators", file=sys.stderr)
        return 1

    result = evaluate_exit(
        indicators=indicators,
        exit_signal=args.exit_signal,
        min_heuristics=args.min_heuristics,
        require_explicit=not args.no_require_explicit
    )

    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print(f"Decision: {'EXIT' if result['should_exit'] else 'CONTINUE'}")
        print(f"Reason: {result['reason']}")
        if result['blocked_reason']:
            print(f"Blocked: {result['blocked_reason']}")

    # Exit code: 0 = should exit, 1 = should continue
    return 0 if result["should_exit"] else 1


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Auto-detect current task phase for Navigator loop mode.

Phases:
- INIT: Loading context, understanding requirements
- RESEARCH: Exploring codebase, finding patterns
- IMPL: Writing code, making changes
- VERIFY: Running tests, validating functionality
- COMPLETE: All indicators met, ready to exit

Usage:
    python3 phase_detector.py \
        --files-read '["src/auth.ts", "README.md"]' \
        --files-changed '["src/login.ts"]' \
        --tests-running \
        --indicators '{"code_committed": true, "tests_passing": true}'

Output:
    JSON with detected phase and confidence
"""

import argparse
import json
import sys
from typing import List, Optional


def detect_phase(
    files_read: List[str],
    files_changed: List[str],
    tests_running: bool = False,
    test_exit_code: Optional[int] = None,
    indicators: dict = None,
    exit_signal: bool = False
) -> dict:
    """
    Auto-detect current task phase from context.

    Returns dict with:
        - phase: str
        - confidence: float (0-1)
        - reason: str
        - next_expected: str
    """
    indicators = indicators or {}
    met_count = sum(1 for v in indicators.values() if v)

    # COMPLETE: Exit conditions met
    if met_count >= 4 and exit_signal:
        return {
            "phase": "COMPLETE",
            "confidence": 1.0,
            "reason": f"Exit conditions met ({met_count}/5 indicators + EXIT_SIGNAL)",
            "next_expected": "Execute autonomous completion protocol"
        }

    # VERIFY: Tests running or recently run
    if tests_running:
        return {
            "phase": "VERIFY",
            "confidence": 0.95,
            "reason": "Tests currently running",
            "next_expected": "Wait for test results, then evaluate"
        }

    if test_exit_code is not None:
        if test_exit_code == 0:
            return {
                "phase": "VERIFY",
                "confidence": 0.9,
                "reason": f"Tests completed (exit code: {test_exit_code})",
                "next_expected": "Commit changes if tests pass"
            }
        else:
            return {
                "phase": "IMPL",
                "confidence": 0.85,
                "reason": f"Tests failed (exit code: {test_exit_code})",
                "next_expected": "Fix failing tests"
            }

    # IMPL: Files being modified
    if files_changed:
        return {
            "phase": "IMPL",
            "confidence": 0.9,
            "reason": f"Files modified: {len(files_changed)} file(s)",
            "next_expected": "Continue implementation or run tests"
        }

    # RESEARCH: Reading files, no changes yet
    if files_read and not files_changed:
        return {
            "phase": "RESEARCH",
            "confidence": 0.85,
            "reason": f"Files read: {len(files_read)} file(s), no changes yet",
            "next_expected": "Start implementation based on research"
        }

    # INIT: Default starting state
    return {
        "phase": "INIT",
        "confidence": 0.7,
        "reason": "No significant activity detected",
        "next_expected": "Load context and understand requirements"
    }


def main():
    parser = argparse.ArgumentParser(
        description="Detect current task phase"
    )
    parser.add_argument("--files-read", default="[]",
                        help="JSON array of files read this iteration")
    parser.add_argument("--files-changed", default="[]",
                        help="JSON array of files changed this iteration")
    parser.add_argument("--tests-running", action="store_true",
                        help="Whether tests are currently running")
    parser.add_argument("--test-exit-code", type=int, default=None,
                        help="Exit code from last test run")
    parser.add_argument("--indicators", default="{}",
                        help="JSON object of completion indicator states")
    parser.add_argument("--exit-signal", action="store_true",
                        help="Whether EXIT_SIGNAL was set")
    parser.add_argument("--init", action="store_true",
                        help="Initialize fresh phase detection")
    parser.add_argument("--output", choices=["json", "text"], default="json",
                        help="Output format")

    args = parser.parse_args()

    # Handle --init flag
    if args.init:
        result = {
            "phase": "INIT",
            "confidence": 1.0,
            "reason": "Fresh initialization",
            "next_expected": "Load context and understand task"
        }
    else:
        try:
            files_read = json.loads(args.files_read)
            files_changed = json.loads(args.files_changed)
            indicators = json.loads(args.indicators)
        except json.JSONDecodeError as e:
            print(f"Error: Invalid JSON input - {e}", file=sys.stderr)
            return 1

        result = detect_phase(
            files_read=files_read,
            files_changed=files_changed,
            tests_running=args.tests_running,
            test_exit_code=args.test_exit_code,
            indicators=indicators,
            exit_signal=args.exit_signal
        )

    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print(f"Phase: {result['phase']}")
        print(f"Confidence: {result['confidence']:.0%}")
        print(f"Reason: {result['reason']}")
        print(f"Next: {result['next_expected']}")

    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Stagnation detection for Navigator loop mode.

Detects when the loop is stuck in the same state by comparing
state hashes across iterations.

Usage:
    python3 stagnation_detector.py \
        --phase "IMPL" \
        --indicators '{"code_committed": false}' \
        --files-changed '["src/auth.ts"]' \
        --history '["abc123", "abc123"]' \
        --threshold 3

Output:
    JSON with stagnation status and hash
"""

import argparse
import hashlib
import json
import sys
from typing import List, Tuple


def calculate_state_hash(
    phase: str,
    indicators: dict,
    files_changed: List[str],
    error_state: str = None
) -> str:
    """
    Generate hash representing current state.

    Same hash = same state = potential stagnation.
    """
    state_components = {
        "phase": phase,
        "indicators": sorted([
            k for k, v in indicators.items() if v
        ]),
        "files_changed": sorted(files_changed) if files_changed else [],
        "error_state": error_state
    }

    state_json = json.dumps(state_components, sort_keys=True)
    return hashlib.md5(state_json.encode()).hexdigest()[:6]


def count_consecutive_same(history: List[str], current_hash: str) -> int:
    """Count consecutive occurrences of current hash in history."""
    consecutive = 1  # Current counts as 1

    for prev_hash in reversed(history):
        if prev_hash == current_hash:
            consecutive += 1
        else:
            break

    return consecutive


def check_stagnation(
    current_hash: str,
    history: List[str],
    threshold: int = 3
) -> Tuple[bool, int]:
    """
    Check if loop is stagnating.

    Returns:
        (is_stagnant, consecutive_count)
    """
    consecutive = count_consecutive_same(history, current_hash)
    is_stagnant = consecutive >= threshold
    return is_stagnant, consecutive


def detect_stagnation(
    phase: str,
    indicators: dict,
    files_changed: List[str],
    history: List[str],
    threshold: int = 3,
    error_state: str = None
) -> dict:
    """
    Full stagnation detection.

    Returns dict with:
        - current_hash: str
        - is_stagnant: bool
        - consecutive_count: int
        - threshold: int
        - recommendation: str
    """
    current_hash = calculate_state_hash(
        phase=phase,
        indicators=indicators,
        files_changed=files_changed,
        error_state=error_state
    )

    is_stagnant, consecutive = check_stagnation(
        current_hash=current_hash,
        history=history,
        threshold=threshold
    )

    # Generate recommendation
    if is_stagnant:
        recommendation = "PAUSE: Same state detected. User intervention needed."
    elif consecutive >= threshold - 1:
        recommendation = "WARNING: Approaching stagnation threshold."
    else:
        recommendation = "OK: State is changing normally."

    return {
        "current_hash": current_hash,
        "previous_hash": history[-1] if history else None,
        "is_stagnant": is_stagnant,
        "consecutive_count": consecutive,
        "threshold": threshold,
        "recommendation": recommendation,
        "state_components": {
            "phase": phase,
            "met_indicators": [k for k, v in indicators.items() if v],
            "files_changed_count": len(files_changed) if files_changed else 0
        }
    }


def main():
    parser = argparse.ArgumentParser(
        description="Detect loop stagnation"
    )
    parser.add_argument("--phase", default="INIT",
                        help="Current phase")
    parser.add_argument("--indicators", default="{}",
                        help="JSON object of indicator states")
    parser.add_argument("--files-changed", default="[]",
                        help="JSON array of changed files")
    parser.add_argument("--history", default="[]",
                        help="JSON array of previous state hashes")
    parser.add_argument("--threshold", type=int, default=3,
                        help="Stagnation threshold (default: 3)")
    parser.add_argument("--error-state", default=None,
                        help="Current error state if any")
    parser.add_argument("--output", choices=["json", "text"], default="json",
                        help="Output format")

    args = parser.parse_args()

    try:
        indicators = json.loads(args.indicators)
        files_changed = json.loads(args.files_changed)
        history = json.loads(args.history)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON input - {e}", file=sys.stderr)
        return 1

    result = detect_stagnation(
        phase=args.phase,
        indicators=indicators,
        files_changed=files_changed,
        history=history,
        threshold=args.threshold,
        error_state=args.error_state
    )

    if args.output == "json":
        print(json.dumps(result, indent=2))
    else:
        print(f"Hash: {result['current_hash']}")
        print(f"Stagnant: {result['is_stagnant']}")
        print(f"Consecutive: {result['consecutive_count']}/{result['threshold']}")
        print(f"Status: {result['recommendation']}")

    # Exit code: 1 = stagnant, 0 = OK
    return 1 if result["is_stagnant"] else 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""
Generate NAVIGATOR_STATUS block for loop mode.

Usage:
    python3 status_generator.py \
        --phase "IMPL" \
        --iteration 2 \
        --max-iterations 5 \
        --indicators '{"code_committed": true, "tests_passing": true}' \
        --state-hash "a7b3c9" \
        --prev-hash "a7b3c9" \
        --stagnation-count 1 \
        --next-action "Run tests"
"""

import argparse
import json
import sys


def calculate_progress(phase: str, indicators: dict) -> int:
    """Calculate progress percentage based on phase and indicators."""
    phase_weights = {
        "INIT": 10,
        "RESEARCH": 25,
        "IMPL": 50,
        "VERIFY": 75,
        "COMPLETE": 100
    }

    base = phase_weights.get(phase, 0)
    indicator_count = sum(1 for v in indicators.values() if v)
    indicator_bonus = (indicator_count / max(len(indicators), 1)) * 25

    return min(100, int(base + indicator_bonus))


def format_indicators(indicators: dict) -> str:
    """Format completion indicators with checkboxes."""
    indicator_labels = {
        "code_committed": "Code changes committed",
        "tests_passing": "Tests passing",
        "docs_updated": "Documentation updated",
        "ticket_closed": "Ticket closed",
        "marker_created": "Marker created"
    }

    lines = []
    for key, label in indicator_labels.items():
        checked = indicators.get(key, False)
        mark = "x" if checked else " "
        lines.append(f"  [{mark}] {label}")

    return "\n".join(lines)


def count_met_indicators(indicators: dict) -> tuple:
    """Count met vs total indicators."""
    met = sum(1 for v in indicators.values() if v)
    total = len(indicators) if indicators else 5
    return met, total


def generate_status_block(
    phase: str,
    iteration: int,
    max_iterations: int,
    indicators: dict,
    state_hash: str,
    prev_hash: str,
    stagnation_count: int,
    stagnation_threshold: int = 3,
    exit_signal: bool = False,
    next_action: str = "Continue working"
) -> str:
    """Generate formatted NAVIGATOR_STATUS block."""

    progress = calculate_progress(phase, indicators)
    indicator_display = format_indicators(indicators)
    met, total = count_met_indicators(indicators)

    status = f"""
NAVIGATOR_STATUS
{'=' * 50}
Phase: {phase}
Iteration: {iteration}/{max_iterations}
Progress: {progress}%

Completion Indicators:
{indicator_display}

Exit Conditions:
  Heuristics: {met}/{total} (need 2+)
  EXIT_SIGNAL: {str(exit_signal).lower()}

State Hash: {state_hash}
Previous Hash: {prev_hash}
Stagnation: {stagnation_count}/{stagnation_threshold}

Next Action: {next_action}
{'=' * 50}
"""
    return status.strip()


def main():
    parser = argparse.ArgumentParser(
        description="Generate NAVIGATOR_STATUS block"
    )
    parser.add_argument("--phase", default="INIT",
                        choices=["INIT", "RESEARCH", "IMPL", "VERIFY", "COMPLETE"])
    parser.add_argument("--iteration", type=int, default=1)
    parser.add_argument("--max-iterations", type=int, default=5)
    parser.add_argument("--indicators", default="{}")
    parser.add_argument("--state-hash", default="000000")
    parser.add_argument("--prev-hash", default="000000")
    parser.add_argument("--stagnation-count", type=int, default=0)
    parser.add_argument("--stagnation-threshold", type=int, default=3)
    parser.add_argument("--exit-signal", action="store_true")
    parser.add_argument("--next-action", default="Continue working")

    args = parser.parse_args()

    try:
        indicators = json.loads(args.indicators)
    except json.JSONDecodeError:
        indicators = {}

    status = generate_status_block(
        phase=args.phase,
        iteration=args.iteration,
        max_iterations=args.max_iterations,
        indicators=indicators,
        state_hash=args.state_hash,
        prev_hash=args.prev_hash,
        stagnation_count=args.stagnation_count,
        stagnation_threshold=args.stagnation_threshold,
        exit_signal=args.exit_signal,
        next_action=args.next_action
    )

    print(status)
    return 0


if __name__ == "__main__":
    sys.exit(main())