Writing Detectors

How to create custom program and state detectors.

Overview

Detectors identify programs running in terminal sessions and determine their current state. Each detector:

1. Checks if a specific program is running
2. Detects the program's state (Idle, Working, Blocked, Error)
3. Returns confidence scores for matches

Detector Interface

BaseDetector

All detectors extend BaseDetector:

from agentwatch.detectors.base import BaseDetector, DetectionResult

class MyDetector(BaseDetector):
    """Detector for MyProgram."""

    @property
    def priority(self) -> int:
        """Higher priority detectors run first.

        Priority guidelines:
        - 100: Very specific program (Claude Code)
        - 50: Specific program (Codex, etc.)
        - 10: Generic fallback (shell)
        """
        return 50

    def detect_program(self, text: str) -> Optional[DetectionResult]:
        """Detect if this program is running.

        Args:
            text: ANSI-stripped terminal content

        Returns:
            DetectionResult with program name and confidence,
            or None if program not detected.
        """
        pass

    def detect_state(self, text: str, program: str) -> Optional[DetectionResult]:
        """Detect current state of the program.

        Args:
            text: ANSI-stripped terminal content
            program: Detected program name

        Returns:
            DetectionResult with state and detail,
            or None if state cannot be determined.
        """
        pass

DetectionResult

@dataclass
class DetectionResult:
    """Result of a detection."""
    name: str           # Program or state name
    confidence: float   # 0.0 to 1.0
    detail: Optional[str] = None  # Additional context
    version: Optional[str] = None  # For programs

Program Detection

Basic Example

class MyDetector(BaseDetector):
    @property
    def priority(self) -> int:
        return 50

    def detect_program(self, text: str) -> Optional[DetectionResult]:
        # Look for distinctive program header
        if "MyProgram v" in text:
            # Extract version
            match = re.search(r'MyProgram v([\d.]+)', text)
            version = match.group(1) if match else None

            return DetectionResult(
                name="MyProgram",
                confidence=1.0,
                version=version
            )

        # Check for secondary indicators
        if "myprogram>" in text:
            return DetectionResult(
                name="MyProgram",
                confidence=0.8
            )

        return None

Confidence Guidelines

Confidence	When to Use
1.0	Definitive match (version header, unique identifier)
0.9	Very strong match (distinctive UI element)
0.7-0.8	Good match (characteristic patterns)
0.5-0.6	Weak match (common elements like prompts)
<0.5	Don't return (threshold not met)

Pattern Matching

Use the patterns module for consistency:

from agentwatch.detectors.patterns import ProgramPattern

MY_PATTERNS = [
    ProgramPattern(
        name="version_header",
        confidence=1.0,
        reason="Version header is definitive",
        regex=r'MyProgram v([\d.]+)',
    ),
    ProgramPattern(
        name="prompt",
        confidence=0.6,
        reason="Prompt is common but characteristic",
        contains="myprogram>",
    ),
]

State Detection

Pattern-Based Detection

Use weighted patterns:

from agentwatch.detectors.patterns import Pattern

IDLE_PATTERNS = [
    Pattern(
        name="prompt_visible",
        weight=50,
        reason="Prompt visible indicates waiting for input",
        contains="myprogram>",
    ),
    Pattern(
        name="welcome_message",
        weight=30,
        reason="Welcome message appears when idle",
        contains="Welcome to MyProgram",
    ),
]

WORKING_PATTERNS = [
    Pattern(
        name="processing",
        weight=80,
        reason="Processing indicator is strong working signal",
        contains="Processing...",
    ),
    Pattern(
        name="spinner",
        weight=50,
        reason="Spinner indicates active work",
        regex=r'[⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏]',
    ),
]

State Detection Implementation

def detect_state(self, text: str, program: str) -> Optional[DetectionResult]:
    if program != "MyProgram":
        return None

    # Calculate scores for each state
    idle_score = sum(p.weight for p in IDLE_PATTERNS if p.matches(text))
    working_score = sum(p.weight for p in WORKING_PATTERNS if p.matches(text))
    blocked_score = sum(p.weight for p in BLOCKED_PATTERNS if p.matches(text))
    error_score = sum(p.weight for p in ERROR_PATTERNS if p.matches(text))

    # Find highest score
    scores = {
        "Idle": idle_score,
        "Working": working_score,
        "Blocked": blocked_score,
        "Error": error_score,
    }
    state = max(scores, key=scores.get)
    max_score = scores[state]

    if max_score == 0:
        return None

    # Calculate confidence based on margin
    total = sum(scores.values())
    confidence = max_score / total if total > 0 else 0

    # Extract detail
    detail = self._extract_detail(text, state)

    return DetectionResult(
        name=state,
        confidence=confidence,
        detail=detail
    )

State Details

Provide context in the detail field:

State	Example Details
Idle	"Ready", "Waiting for input"
Working	"Processing", "Running command", "Thinking"
Blocked	"Permission dialog", "Menu selection", "Confirmation"
Error	"Connection failed", "Rate limited", "Command failed"

Registration

Register your detector in agentwatch/detectors/__init__.py:

from .my_detector import MyDetector

# In get_all_detectors():
return [
    ClaudeCodeDetector(),
    MyDetector(),  # Add here
    ShellDetector(),  # Keep shell last (lowest priority)
]

Testing

Test Structure

Create agentwatch/tests/test_detectors_my_program.py:

import pytest
from agentwatch.detectors.my_detector import MyDetector

class TestMyProgramDetection:
    """Tests for MyProgram program detection."""

    @pytest.fixture
    def detector(self):
        return MyDetector()

    def test_detects_with_version_header(self, detector):
        text = """
        MyProgram v1.2.3
        Ready for input
        myprogram>
        """
        result = detector.detect_program(text)
        assert result is not None
        assert result.name == "MyProgram"
        assert result.confidence == 1.0
        assert result.version == "1.2.3"

    def test_detects_without_version(self, detector):
        text = """
        myprogram> hello
        Processing...
        """
        result = detector.detect_program(text)
        assert result is not None
        assert result.confidence < 1.0


class TestMyProgramStateDetection:
    """Tests for MyProgram state detection."""

    @pytest.fixture
    def detector(self):
        return MyDetector()

    def test_idle_at_prompt(self, detector):
        text = """
        MyProgram v1.2.3
        Welcome to MyProgram
        myprogram>
        """
        result = detector.detect_state(text, "MyProgram")
        assert result is not None
        assert result.name == "Idle"

    def test_working_when_processing(self, detector):
        text = """
        MyProgram v1.2.3
        Processing... ⠹
        """
        result = detector.detect_state(text, "MyProgram")
        assert result is not None
        assert result.name == "Working"

Using Real Captures

Save real terminal captures as test fixtures:

# Capture a real session
tmux capture-pane -p -t mysession > test_case.txt

# Move to test fixtures
cp test_case.txt agentwatch/tests/fixtures/my_program_idle.txt

Use in tests:

@pytest.fixture
def idle_capture(self):
    with open("tests/fixtures/my_program_idle.txt") as f:
        return f.read()

def test_idle_from_real_capture(self, detector, idle_capture):
    result = detector.detect_state(idle_capture, "MyProgram")
    assert result.name == "Idle"

Best Practices

Pattern Design

1. Be specific: Prefer unique patterns over generic ones
2. Use weights: Higher weights for more reliable signals
3. Document reasoning: Explain why patterns have their weights
4. Handle edge cases: Consider partial matches, truncated output

Confidence Scores

1. Definitive matches: 1.0 for version headers, unique identifiers
2. Strong matches: 0.8-0.9 for distinctive UI elements
3. Weak matches: 0.5-0.7 for common patterns
4. No match: Return None, don't return low confidence

State Priority

When patterns conflict:

1. Error patterns should have high weights (errors are serious)
2. Blocked should override Idle (waiting for action is more important)
3. Working should override Idle when active indicators present
4. Use timing information when available (stale indicators decay)

Performance

1. Simple patterns first: Check cheap patterns before expensive regex
2. Short-circuit: Return early on definitive matches
3. Cache compiled regex: Use module-level compiled patterns

Example: Full Detector

"""Detector for MyProgram."""

import re
from typing import Optional
from .base import BaseDetector, DetectionResult
from .patterns import Pattern

# Program patterns
VERSION_PATTERN = re.compile(r'MyProgram v([\d.]+)')
PROMPT_PATTERN = re.compile(r'^myprogram>\s*$', re.MULTILINE)

# State patterns
IDLE_PATTERNS = [
    Pattern("prompt_visible", 50, "Prompt visible", contains="myprogram>"),
    Pattern("welcome", 30, "Welcome message", contains="Welcome"),
]

WORKING_PATTERNS = [
    Pattern("processing", 80, "Processing indicator", contains="Processing"),
    Pattern("spinner", 50, "Spinner animation", regex=r'[⠋⠙⠹⠸⠼⠴⠦⠧⠇⠏]'),
]

BLOCKED_PATTERNS = [
    Pattern("confirm", 90, "Confirmation prompt", contains="[y/n]"),
    Pattern("menu", 70, "Menu selection", regex=r'^\d+\.\s+\w', re.MULTILINE),
]

ERROR_PATTERNS = [
    Pattern("error_message", 80, "Error keyword", regex=r'^Error:', re.MULTILINE),
]


class MyDetector(BaseDetector):
    """Detector for MyProgram sessions."""

    @property
    def priority(self) -> int:
        return 50

    def detect_program(self, text: str) -> Optional[DetectionResult]:
        # Check for version header (definitive)
        match = VERSION_PATTERN.search(text)
        if match:
            return DetectionResult(
                name="MyProgram",
                confidence=1.0,
                version=match.group(1)
            )

        # Check for prompt (weaker)
        if PROMPT_PATTERN.search(text):
            return DetectionResult(
                name="MyProgram",
                confidence=0.6
            )

        return None

    def detect_state(self, text: str, program: str) -> Optional[DetectionResult]:
        if program != "MyProgram":
            return None

        scores = {
            "Idle": self._score_patterns(text, IDLE_PATTERNS),
            "Working": self._score_patterns(text, WORKING_PATTERNS),
            "Blocked": self._score_patterns(text, BLOCKED_PATTERNS),
            "Error": self._score_patterns(text, ERROR_PATTERNS),
        }

        state = max(scores, key=scores.get)
        if scores[state] == 0:
            return DetectionResult(name="Idle", confidence=0.5)

        total = sum(scores.values())
        confidence = scores[state] / total

        return DetectionResult(
            name=state,
            confidence=confidence,
            detail=self._get_detail(text, state)
        )

    def _score_patterns(self, text: str, patterns: list) -> int:
        return sum(p.weight for p in patterns if p.matches(text))

    def _get_detail(self, text: str, state: str) -> Optional[str]:
        if state == "Working":
            if "Processing" in text:
                return "Processing"
        elif state == "Blocked":
            if "[y/n]" in text:
                return "Confirmation"
        elif state == "Error":
            match = re.search(r'Error:\s*(.+)', text)
            if match:
                return match.group(1)[:50]
        return None