preflight

preflight

Pre-flight checks and prompt analysis before sheet execution.

Analyzes prompts before execution to estimate token usage, extract file references, and detect potential issues early.

Classes

PromptMetrics dataclass

PromptMetrics(character_count, estimated_tokens, line_count, has_file_references, referenced_paths=list(), word_count=0)

Metrics about a prompt for monitoring and optimization.

Provides visibility into prompt size and complexity before execution.

Attributes

character_count instance-attribute

character_count

Total character count in the prompt.

estimated_tokens instance-attribute

estimated_tokens

Estimated token count via centralized estimator (tokens.estimate_tokens).

line_count instance-attribute

line_count

Number of lines in the prompt.

has_file_references instance-attribute

has_file_references

Whether the prompt contains file path references.

referenced_paths class-attribute instance-attribute

referenced_paths = field(default_factory=list)

List of file paths extracted from the prompt.

word_count class-attribute instance-attribute

word_count = 0

Approximate word count for additional context.

Functions

from_prompt classmethod

from_prompt(prompt)

Analyze a prompt and extract metrics.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt text to analyze.	required

Returns:

Type	Description
PromptMetrics	PromptMetrics with extracted data.

Source code in src/marianne/execution/preflight.py

@classmethod
def from_prompt(cls, prompt: str) -> "PromptMetrics":
    """Analyze a prompt and extract metrics.

    Args:
        prompt: The prompt text to analyze.

    Returns:
        PromptMetrics with extracted data.
    """
    character_count = len(prompt)
    line_count = prompt.count("\n") + 1
    estimated_tokens = estimate_tokens(prompt)
    word_count = len(prompt.split())

    # Extract file references
    referenced_paths = cls._extract_file_paths(prompt)
    has_file_references = len(referenced_paths) > 0

    return cls(
        character_count=character_count,
        estimated_tokens=estimated_tokens,
        line_count=line_count,
        has_file_references=has_file_references,
        referenced_paths=referenced_paths,
        word_count=word_count,
    )

to_dict

to_dict()

Convert to serializable dictionary.

Source code in src/marianne/execution/preflight.py

def to_dict(self) -> PromptMetricsDict:
    """Convert to serializable dictionary."""
    # Suppress return-value: mypy/pyright cannot infer a dict literal
    # as matching a total=False TypedDict — all keys verified via schema
    return {
        "character_count": self.character_count,
        "estimated_tokens": self.estimated_tokens,
        "line_count": self.line_count,
        "word_count": self.word_count,
        "has_file_references": self.has_file_references,
        "referenced_paths": self.referenced_paths,
    }

PreflightResult dataclass

PreflightResult(prompt_metrics, warnings=list(), errors=list(), paths_accessible=dict(), working_directory_valid=True)

Result of pre-flight checks before sheet execution.

Captures both prompt metrics and any warnings/errors that should be addressed before or during execution.

Attributes

prompt_metrics instance-attribute

prompt_metrics

Metrics about the prompt being executed.

warnings class-attribute instance-attribute

warnings = field(default_factory=list)

Non-fatal warnings that should be logged but don't block execution.

errors class-attribute instance-attribute

errors = field(default_factory=list)

Fatal issues that should prevent execution.

paths_accessible class-attribute instance-attribute

paths_accessible = field(default_factory=dict)

Mapping of referenced paths to their accessibility status.

working_directory_valid class-attribute instance-attribute

working_directory_valid = True

Whether the working directory exists and is accessible.

has_errors property

has_errors

Check if there are any fatal errors.

has_warnings property

has_warnings

Check if there are any warnings.

can_proceed property

can_proceed

Check if execution can proceed (no fatal errors).

inaccessible_paths property

inaccessible_paths

Get list of referenced paths that don't exist.

Functions

to_dict

to_dict()

Convert to serializable dictionary.

Source code in src/marianne/execution/preflight.py

def to_dict(self) -> dict[str, Any]:
    """Convert to serializable dictionary."""
    return {
        "prompt_metrics": self.prompt_metrics.to_dict(),
        "warnings": self.warnings,
        "errors": self.errors,
        "paths_accessible": self.paths_accessible,
        "working_directory_valid": self.working_directory_valid,
        "can_proceed": self.can_proceed,
    }

PreflightChecker

PreflightChecker(workspace, working_directory=None, token_warning_threshold=TOKEN_WARNING_THRESHOLD, token_error_threshold=TOKEN_ERROR_THRESHOLD)

Performs pre-flight checks before sheet execution.

Analyzes prompts for potential issues, verifies file references, and provides warnings for resource-intensive prompts.

Initialize preflight checker.

Parameters:

Name	Type	Description	Default
workspace	Path	Base workspace directory for resolving relative paths.	required
working_directory	Path | None	Working directory for execution (defaults to workspace).	None
token_warning_threshold	int	Token count above which to warn.	TOKEN_WARNING_THRESHOLD
token_error_threshold	int	Token count above which to error.	TOKEN_ERROR_THRESHOLD

Source code in src/marianne/execution/preflight.py

def __init__(
    self,
    workspace: Path,
    working_directory: Path | None = None,
    token_warning_threshold: int = TOKEN_WARNING_THRESHOLD,
    token_error_threshold: int = TOKEN_ERROR_THRESHOLD,
) -> None:
    """Initialize preflight checker.

    Args:
        workspace: Base workspace directory for resolving relative paths.
        working_directory: Working directory for execution (defaults to workspace).
        token_warning_threshold: Token count above which to warn.
        token_error_threshold: Token count above which to error.
    """
    self.workspace = workspace.resolve()
    self.working_directory = (working_directory or workspace).resolve()
    self.token_warning_threshold = token_warning_threshold
    self.token_error_threshold = token_error_threshold

Functions

check

check(prompt, sheet_context=None)

Perform pre-flight checks on a prompt.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt text to analyze.	required
sheet_context	dict[str, Any] | None	Optional context for template variable expansion.	None

Returns:

Type	Description
PreflightResult	PreflightResult with metrics, warnings, and errors.

Source code in src/marianne/execution/preflight.py

def check(
    self,
    prompt: str,
    sheet_context: dict[str, Any] | None = None,
) -> PreflightResult:
    """Perform pre-flight checks on a prompt.

    Args:
        prompt: The prompt text to analyze.
        sheet_context: Optional context for template variable expansion.

    Returns:
        PreflightResult with metrics, warnings, and errors.
    """
    # Analyze prompt metrics
    metrics = PromptMetrics.from_prompt(prompt)

    warnings: list[str] = []
    errors: list[str] = []
    paths_accessible: dict[str, bool] = {}

    # Check token count thresholds
    if metrics.estimated_tokens > self.token_error_threshold:
        errors.append(
            f"Prompt exceeds token limit: ~{metrics.estimated_tokens:,} tokens "
            f"(threshold: {self.token_error_threshold:,}). "
            "Consider reducing prompt size to avoid context window issues."
        )
    elif metrics.estimated_tokens > self.token_warning_threshold:
        warnings.append(
            f"Large prompt detected: ~{metrics.estimated_tokens:,} tokens "
            f"(warning threshold: {self.token_warning_threshold:,}). "
            "This may impact response quality or cause truncation."
        )

    # Check line count
    if metrics.line_count > LINE_WARNING_THRESHOLD:
        warnings.append(
            f"Prompt has many lines: {metrics.line_count:,} lines. "
            "Very long prompts may reduce output quality."
        )

    # Verify working directory
    working_directory_valid = (
        self.working_directory.exists()
        and self.working_directory.is_dir()
    )
    if not working_directory_valid:
        errors.append(
            f"Working directory does not exist or is not accessible: "
            f"{self.working_directory}"
        )

    # Verify referenced file paths
    if metrics.has_file_references:
        paths_accessible = self._check_paths(
            metrics.referenced_paths,
            sheet_context or {},
        )
        inaccessible = [p for p, exists in paths_accessible.items() if not exists]
        if inaccessible:
            # This is a warning, not an error - files might be created by the prompt
            warnings.append(
                f"Some referenced paths do not exist: {', '.join(inaccessible[:5])}"
                + (f" (+{len(inaccessible) - 5} more)" if len(inaccessible) > 5 else "")
            )

    return PreflightResult(
        prompt_metrics=metrics,
        warnings=warnings,
        errors=errors,
        paths_accessible=paths_accessible,
        working_directory_valid=working_directory_valid,
    )

Functions

run_preflight_check

run_preflight_check(prompt, workspace, working_directory=None, sheet_context=None)

Convenience function to run preflight checks.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt text to analyze.	required
workspace	Path	Base workspace directory.	required
working_directory	Path | None	Optional working directory for execution.	None
sheet_context	dict[str, Any] | None	Optional context for template expansion.	None

Returns:

Type	Description
PreflightResult	PreflightResult with metrics and any warnings/errors.

Source code in src/marianne/execution/preflight.py

def run_preflight_check(
    prompt: str,
    workspace: Path,
    working_directory: Path | None = None,
    sheet_context: dict[str, Any] | None = None,
) -> PreflightResult:
    """Convenience function to run preflight checks.

    Args:
        prompt: The prompt text to analyze.
        workspace: Base workspace directory.
        working_directory: Optional working directory for execution.
        sheet_context: Optional context for template expansion.

    Returns:
        PreflightResult with metrics and any warnings/errors.
    """
    checker = PreflightChecker(
        workspace=workspace,
        working_directory=working_directory,
    )
    return checker.check(prompt, sheet_context)