escalation

escalation

Escalation protocol for low-confidence sheet execution decisions.

Provides a mechanism for escalating to external decision-makers (human or AI) when sheet confidence is too low to proceed automatically.

Phase 2 of AGI Evolution: Confidence-Based Execution

v21 Evolution: Proactive Checkpoint System - adds pre-execution checkpoints that ask for confirmation BEFORE dangerous operations, complementing the reactive escalation system.

Classes

CheckpointTrigger dataclass

CheckpointTrigger(name, sheet_nums=None, prompt_contains=None, min_retry_count=None, requires_confirmation=True, message='')

Configuration for a proactive checkpoint trigger.

Defines conditions that should trigger a pre-execution checkpoint, asking for confirmation BEFORE the sheet executes.

v21 Evolution: Proactive Checkpoint System - enables pre-action validation.

Attributes

name instance-attribute

name

Name/identifier for this trigger.

sheet_nums class-attribute instance-attribute

sheet_nums = None

Specific sheet numbers to checkpoint (None = all sheets).

prompt_contains class-attribute instance-attribute

prompt_contains = None

Keywords in prompt that trigger checkpoint (case-insensitive).

min_retry_count class-attribute instance-attribute

min_retry_count = None

Trigger if retry count >= this value.

requires_confirmation class-attribute instance-attribute

requires_confirmation = True

Whether to require explicit confirmation (True) or just warn (False).

message class-attribute instance-attribute

message = ''

Custom message to show when checkpoint triggers.

CheckpointContext dataclass

CheckpointContext(job_id, sheet_num, prompt, trigger, retry_count=0, previous_errors=list())

Context provided to checkpoint handlers for decision-making.

Contains all relevant information about the sheet that's about to execute, enabling informed pre-execution decisions.

v21 Evolution: Proactive Checkpoint System.

Attributes

job_id instance-attribute

job_id

Unique identifier for the job.

sheet_num instance-attribute

sheet_num

Sheet number about to execute.

prompt instance-attribute

prompt

The prompt that will be used for sheet execution.

trigger instance-attribute

trigger

The trigger that caused this checkpoint.

retry_count class-attribute instance-attribute

retry_count = 0

Number of retry attempts already made for this sheet.

previous_errors class-attribute instance-attribute

previous_errors = field(default_factory=list)

Errors from previous attempts (if any).

CheckpointResponse dataclass

CheckpointResponse(action, modified_prompt=None, guidance=None)

Response from a checkpoint handler specifying how to proceed.

v21 Evolution: Proactive Checkpoint System.

Attributes

action instance-attribute

action

Action to take: - proceed: Continue with execution - abort: Stop the entire job - skip: Skip this sheet and continue to the next - modify_prompt: Proceed with modified prompt

modified_prompt class-attribute instance-attribute

modified_prompt = None

Modified prompt to use if action is 'modify_prompt'.

guidance class-attribute instance-attribute

guidance = None

Optional guidance or reasoning for the decision.

HistoricalSuggestion dataclass

HistoricalSuggestion(action, outcome, confidence, validation_pass_rate, guidance)

A suggestion from historical escalation decisions.

Represents a past escalation decision that was similar to the current situation, providing guidance based on what worked (or didn't) before.

Attributes

action instance-attribute

action

Action taken in the past (retry, skip, abort, modify_prompt).

outcome instance-attribute

outcome

Outcome after the action (success, failed, skipped, aborted, unknown).

confidence instance-attribute

confidence

Confidence level at which this past escalation occurred.

validation_pass_rate instance-attribute

validation_pass_rate

Validation pass rate at which this past escalation occurred.

guidance instance-attribute

guidance

Any guidance or notes recorded with this past decision.

EscalationContext dataclass

EscalationContext(job_id, sheet_num, validation_results, confidence, retry_count, error_history, prompt_used, output_summary, historical_suggestions=list())

Context provided to escalation handlers for decision-making.

Contains all relevant information about the sheet execution state that led to escalation.

v15 Evolution: Added historical_suggestions field to surface similar past escalation decisions that may inform the current decision.

Attributes

job_id instance-attribute

job_id

Unique identifier for the job.

sheet_num instance-attribute

sheet_num

Sheet number that triggered escalation.

validation_results instance-attribute

validation_results

Serialized validation results from the sheet.

confidence instance-attribute

confidence

Aggregate confidence score that triggered escalation (0.0-1.0).

retry_count instance-attribute

retry_count

Number of retry attempts already made.

error_history instance-attribute

error_history

List of error messages from previous attempts.

prompt_used instance-attribute

prompt_used

The prompt that was used for the sheet execution.

output_summary instance-attribute

output_summary

Summary of the sheet execution output.

historical_suggestions class-attribute instance-attribute

historical_suggestions = field(default_factory=list)

Similar past escalation decisions that may inform this decision.

v15 Evolution: Populated from get_similar_escalation() results. Ordered by relevance (successful outcomes first).

EscalationResponse dataclass

EscalationResponse(action, modified_prompt=None, guidance=None, confidence_boost=0.0)

Response from an escalation handler specifying how to proceed.

Determines the next action for a sheet that triggered escalation.

Attributes

action instance-attribute

action

Action to take: - retry: Retry the sheet with the same or modified prompt - skip: Skip this sheet and continue to the next - abort: Stop the entire job - modify_prompt: Retry with a modified prompt (requires modified_prompt)

modified_prompt class-attribute instance-attribute

modified_prompt = None

Modified prompt to use if action is 'modify_prompt'.

guidance class-attribute instance-attribute

guidance = None

Optional guidance or reasoning for the decision.

confidence_boost class-attribute instance-attribute

confidence_boost = 0.0

Amount to boost confidence threshold for next attempt (0.0-1.0). Allows temporary threshold adjustment based on escalation feedback.

ConsoleEscalationHandler

ConsoleEscalationHandler(confidence_threshold=0.6, auto_retry_on_first_failure=True)

Console-based escalation handler that prompts the user.

Implements the EscalationHandler protocol for interactive human-in-the-loop decision making.

Initialize the console escalation handler.

Parameters:

Name	Type	Description	Default
confidence_threshold	float	Confidence below which to escalate (0.0-1.0).	0.6
auto_retry_on_first_failure	bool	If True, auto-retry on first failure without escalating (reduces noise for transient issues).	True

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

def __init__(
    self,
    confidence_threshold: float = 0.6,
    auto_retry_on_first_failure: bool = True,
) -> None:
    """Initialize the console escalation handler.

    Args:
        confidence_threshold: Confidence below which to escalate (0.0-1.0).
        auto_retry_on_first_failure: If True, auto-retry on first failure
            without escalating (reduces noise for transient issues).
    """
    self.confidence_threshold = confidence_threshold
    self.auto_retry_on_first_failure = auto_retry_on_first_failure

Functions

should_escalate async

should_escalate(sheet_state, validation_result, confidence)

Determine if escalation is needed.

Escalates when: - Confidence is below threshold - AND (not first attempt OR auto_retry_on_first_failure is False)

Parameters:

Name	Type	Description	Default
sheet_state	SheetState	Current sheet state.	required
validation_result	SheetValidationResult	Validation results.	required
confidence	float	Aggregate confidence score.	required

Returns:

Type	Description
bool	True if user should be prompted for a decision.

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

async def should_escalate(
    self,
    sheet_state: SheetState,
    validation_result: SheetValidationResult,
    confidence: float,
) -> bool:
    """Determine if escalation is needed.

    Escalates when:
    - Confidence is below threshold
    - AND (not first attempt OR auto_retry_on_first_failure is False)

    Args:
        sheet_state: Current sheet state.
        validation_result: Validation results.
        confidence: Aggregate confidence score.

    Returns:
        True if user should be prompted for a decision.
    """
    # Don't escalate if confidence is acceptable
    if confidence >= self.confidence_threshold:
        return False

    # Escalate if auto-retry is disabled OR this is not the first attempt
    return not (self.auto_retry_on_first_failure and sheet_state.attempt_count <= 1)

escalate async

escalate(context)

Prompt user for escalation decision via console.

Parameters:

Name	Type	Description	Default
context	EscalationContext	Full escalation context.	required

Returns:

Type	Description
EscalationResponse	EscalationResponse based on user input.

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

async def escalate(self, context: EscalationContext) -> EscalationResponse:
    """Prompt user for escalation decision via console.

    Args:
        context: Full escalation context.

    Returns:
        EscalationResponse based on user input.
    """
    self._print_context_summary(context)
    return await self._prompt_for_action(context)

ConsoleCheckpointHandler

Console-based checkpoint handler that prompts the user before execution.

Implements the CheckpointHandler protocol for interactive human-in-the-loop decision making BEFORE sheet execution.

v21 Evolution: Proactive Checkpoint System.

Functions

should_checkpoint async

should_checkpoint(sheet_num, prompt, retry_count, triggers)

Determine if a checkpoint is needed before sheet execution.

Checks all configured triggers against the current sheet context.

Parameters:

Name	Type	Description	Default
sheet_num	int	Sheet number about to execute.	required
prompt	str	The prompt to be used.	required
retry_count	int	Number of retry attempts already made.	required
triggers	list[CheckpointTrigger]	List of configured checkpoint triggers.	required

Returns:

Type	Description
CheckpointTrigger | None	The first matching CheckpointTrigger, or None if no checkpoint needed.

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

async def should_checkpoint(
    self,
    sheet_num: int,
    prompt: str,
    retry_count: int,
    triggers: list[CheckpointTrigger],
) -> CheckpointTrigger | None:
    """Determine if a checkpoint is needed before sheet execution.

    Checks all configured triggers against the current sheet context.

    Args:
        sheet_num: Sheet number about to execute.
        prompt: The prompt to be used.
        retry_count: Number of retry attempts already made.
        triggers: List of configured checkpoint triggers.

    Returns:
        The first matching CheckpointTrigger, or None if no checkpoint needed.
    """
    prompt_lower = prompt.lower()

    for trigger in triggers:
        # Check sheet number match
        if trigger.sheet_nums is not None and sheet_num not in trigger.sheet_nums:
            continue

        # Check prompt keyword match
        if trigger.prompt_contains is not None:
            keyword_match = any(
                kw.lower() in prompt_lower
                for kw in trigger.prompt_contains
            )
            if not keyword_match:
                continue

        # Check retry count threshold
        if trigger.min_retry_count is not None and retry_count < trigger.min_retry_count:
            continue

        # All conditions passed - trigger matched
        return trigger

    return None

checkpoint async

checkpoint(context)

Handle checkpoint and prompt user for decision.

Parameters:

Name	Type	Description	Default
context	CheckpointContext	Full context about the checkpoint trigger.	required

Returns:

Type	Description
CheckpointResponse	CheckpointResponse with the action to take.

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

async def checkpoint(self, context: CheckpointContext) -> CheckpointResponse:
    """Handle checkpoint and prompt user for decision.

    Args:
        context: Full context about the checkpoint trigger.

    Returns:
        CheckpointResponse with the action to take.
    """
    self._print_checkpoint_prompt(context)

    if not context.trigger.requires_confirmation:
        # Warning only - proceed automatically
        return CheckpointResponse(
            action="proceed",
            guidance="Auto-proceed (warning-only checkpoint)",
        )

    return await self._prompt_for_checkpoint_action(context)