escalation

escalation

Escalation mixin for the global learning store.

This module contains the EscalationMixin class that provides escalation decision recording and learning functionality. When sheets trigger escalation and receive responses from human or AI handlers, this module records the decisions for future pattern-based suggestions.

Evolution v11: Escalation Learning Loop - closes the loop between escalation handlers and learning system, enabling pattern-based suggestions for similar escalation contexts.

Extracted from global_store.py as part of the modularization effort.

Attributes

Classes

EscalationMixin

Mixin providing escalation decision functionality.

This mixin provides methods for recording and querying escalation decisions. When a sheet triggers escalation and receives a response, the decision is recorded so Marianne can learn from it and potentially suggest similar actions for future escalations with similar contexts.

Requires the following from the composed class

• _get_connection() -> context manager yielding sqlite3.Connection
• hash_job(job_id: str) -> str (static method)

Functions

record_escalation_decision

record_escalation_decision(job_id, sheet_num, confidence, action, validation_pass_rate, retry_count, guidance=None, outcome_after_action=None, model=None)

Record an escalation decision for learning.

When a sheet triggers escalation and receives a response from a human or AI handler, this method records the decision so that Marianne can learn from it and potentially suggest similar actions for future escalations with similar contexts.

Evolution v11: Escalation Learning Loop - closes the loop between escalation handlers and learning system.

Parameters:

Name	Type	Description	Default
job_id	str	ID of the job that triggered escalation.	required
sheet_num	int	Sheet number that triggered escalation.	required
confidence	float	Aggregate confidence score at escalation time (0.0-1.0).	required
action	str	Action taken (retry, skip, abort, modify_prompt).	required
validation_pass_rate	float	Pass percentage at escalation time.	required
retry_count	int	Number of retries before escalation.	required
guidance	str | None	Optional guidance/notes from the handler.	None
outcome_after_action	str | None	What happened after (success, failed, etc.).	None
model	str | None	Optional model name used for execution.	None

Returns:

Type	Description
str	The escalation decision record ID.

Source code in src/marianne/learning/store/escalation.py

def record_escalation_decision(
    self,
    job_id: str,
    sheet_num: int,
    confidence: float,
    action: str,
    validation_pass_rate: float,
    retry_count: int,
    guidance: str | None = None,
    outcome_after_action: str | None = None,
    model: str | None = None,
) -> str:
    """Record an escalation decision for learning.

    When a sheet triggers escalation and receives a response from
    a human or AI handler, this method records the decision so that
    Marianne can learn from it and potentially suggest similar actions
    for future escalations with similar contexts.

    Evolution v11: Escalation Learning Loop - closes the loop between
    escalation handlers and learning system.

    Args:
        job_id: ID of the job that triggered escalation.
        sheet_num: Sheet number that triggered escalation.
        confidence: Aggregate confidence score at escalation time (0.0-1.0).
        action: Action taken (retry, skip, abort, modify_prompt).
        validation_pass_rate: Pass percentage at escalation time.
        retry_count: Number of retries before escalation.
        guidance: Optional guidance/notes from the handler.
        outcome_after_action: What happened after (success, failed, etc.).
        model: Optional model name used for execution.

    Returns:
        The escalation decision record ID.
    """
    record_id = str(uuid.uuid4())
    job_hash = self.hash_job(job_id)
    now = datetime.now()

    with self._get_connection() as conn:
        conn.execute(
            """
            INSERT INTO escalation_decisions (
                id, job_hash, sheet_num, confidence, action,
                guidance, validation_pass_rate, retry_count,
                outcome_after_action, recorded_at, model
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                record_id,
                job_hash,
                sheet_num,
                confidence,
                action,
                guidance,
                validation_pass_rate,
                retry_count,
                outcome_after_action,
                now.isoformat(),
                model,
            ),
        )

    _logger.info(
        f"Recorded escalation decision {record_id}: sheet={sheet_num}, "
        f"action={action}, confidence={confidence:.1%}"
    )
    return record_id

get_escalation_history

get_escalation_history(job_id=None, action=None, limit=20)

Get historical escalation decisions.

Retrieves past escalation decisions for analysis or display. Can filter by job or action type.

Parameters:

Name	Type	Description	Default
job_id	str | None	Optional job ID to filter by.	None
action	str | None	Optional action type to filter by.	None
limit	int	Maximum number of records to return.	20

Returns:

Type	Description
list[EscalationDecisionRecord]	List of EscalationDecisionRecord objects.

Source code in src/marianne/learning/store/escalation.py

def get_escalation_history(
    self,
    job_id: str | None = None,
    action: str | None = None,
    limit: int = 20,
) -> list[EscalationDecisionRecord]:
    """Get historical escalation decisions.

    Retrieves past escalation decisions for analysis or display.
    Can filter by job or action type.

    Args:
        job_id: Optional job ID to filter by.
        action: Optional action type to filter by.
        limit: Maximum number of records to return.

    Returns:
        List of EscalationDecisionRecord objects.
    """
    with self._get_connection() as conn:
        wb = WhereBuilder()
        if job_id is not None:
            wb.add("job_hash = ?", self.hash_job(job_id))
        if action is not None:
            wb.add("action = ?", action)

        where_sql, params = wb.build()
        cursor = conn.execute(
            f"""
            SELECT * FROM escalation_decisions
            WHERE {where_sql}
            ORDER BY recorded_at DESC
            LIMIT ?
            """,
            (*params, limit),
        )

        records = []
        for row in cursor.fetchall():
            records.append(
                EscalationDecisionRecord(
                    id=row["id"],
                    job_hash=row["job_hash"],
                    sheet_num=row[SHEET_NUM_KEY],
                    confidence=row["confidence"],
                    action=row["action"],
                    guidance=row["guidance"],
                    validation_pass_rate=row[VALIDATION_PASS_RATE_KEY],
                    retry_count=row["retry_count"],
                    outcome_after_action=row["outcome_after_action"],
                    recorded_at=datetime.fromisoformat(row["recorded_at"]),
                    model=row["model"],
                )
            )

        return records

get_similar_escalation

get_similar_escalation(confidence, validation_pass_rate, confidence_tolerance=0.15, pass_rate_tolerance=15.0, limit=5)

Get similar past escalation decisions for guidance.

Finds historical escalations with similar context (confidence and pass rate) to help inform the current escalation decision. Can be used to suggest actions or provide guidance to human operators.

Evolution v11: Escalation Learning Loop - enables pattern-based suggestions for similar escalation contexts.

Parameters:

Name	Type	Description	Default
confidence	float	Current confidence level (0.0-1.0).	required
validation_pass_rate	float	Current validation pass percentage.	required
confidence_tolerance	float	How much confidence can differ (default 0.15).	0.15
pass_rate_tolerance	float	How much pass rate can differ (default 15%).	15.0
limit	int	Maximum number of similar records to return.	5

Returns:

Type	Description
list[EscalationDecisionRecord]	List of EscalationDecisionRecord from similar past escalations,
list[EscalationDecisionRecord]	ordered by outcome success (successful outcomes first).

Source code in src/marianne/learning/store/escalation.py

def get_similar_escalation(
    self,
    confidence: float,
    validation_pass_rate: float,
    confidence_tolerance: float = 0.15,
    pass_rate_tolerance: float = 15.0,
    limit: int = 5,
) -> list[EscalationDecisionRecord]:
    """Get similar past escalation decisions for guidance.

    Finds historical escalations with similar context (confidence and
    pass rate) to help inform the current escalation decision. Can be
    used to suggest actions or provide guidance to human operators.

    Evolution v11: Escalation Learning Loop - enables pattern-based
    suggestions for similar escalation contexts.

    Args:
        confidence: Current confidence level (0.0-1.0).
        validation_pass_rate: Current validation pass percentage.
        confidence_tolerance: How much confidence can differ (default 0.15).
        pass_rate_tolerance: How much pass rate can differ (default 15%).
        limit: Maximum number of similar records to return.

    Returns:
        List of EscalationDecisionRecord from similar past escalations,
        ordered by outcome success (successful outcomes first).
    """
    with self._get_connection() as conn:
        # Find escalations with similar confidence and pass rate
        # Order by: successful outcomes first, then by how close the match is
        cursor = conn.execute(
            """
            SELECT *,
                   ABS(confidence - ?) as conf_diff,
                   ABS(validation_pass_rate - ?) as rate_diff
            FROM escalation_decisions
            WHERE ABS(confidence - ?) <= ?
              AND ABS(validation_pass_rate - ?) <= ?
            ORDER BY
                CASE WHEN outcome_after_action = 'success' THEN 0
                     WHEN outcome_after_action = 'skipped' THEN 1
                     WHEN outcome_after_action IS NULL THEN 2
                     ELSE 3 END,
                conf_diff + (rate_diff / 100.0)
            LIMIT ?
            """,
            (
                confidence,
                validation_pass_rate,
                confidence,
                confidence_tolerance,
                validation_pass_rate,
                pass_rate_tolerance,
                limit,
            ),
        )

        records = []
        for row in cursor.fetchall():
            records.append(
                EscalationDecisionRecord(
                    id=row["id"],
                    job_hash=row["job_hash"],
                    sheet_num=row[SHEET_NUM_KEY],
                    confidence=row["confidence"],
                    action=row["action"],
                    guidance=row["guidance"],
                    validation_pass_rate=row[VALIDATION_PASS_RATE_KEY],
                    retry_count=row["retry_count"],
                    outcome_after_action=row["outcome_after_action"],
                    recorded_at=datetime.fromisoformat(row["recorded_at"]),
                    model=row["model"],
                )
            )

        return records

update_escalation_outcome

update_escalation_outcome(escalation_id, outcome_after_action)

Update the outcome of an escalation decision.

Called after an escalation action is taken and the result is known. This closes the feedback loop by recording whether the action led to success or failure.

Parameters:

Name	Type	Description	Default
escalation_id	str	The escalation record ID to update.	required
outcome_after_action	str	What happened (success, failed, aborted, skipped).	required

Returns:

Type	Description
bool	True if the record was updated, False if not found.

Source code in src/marianne/learning/store/escalation.py

def update_escalation_outcome(
    self,
    escalation_id: str,
    outcome_after_action: str,
) -> bool:
    """Update the outcome of an escalation decision.

    Called after an escalation action is taken and the result is known.
    This closes the feedback loop by recording whether the action led
    to success or failure.

    Args:
        escalation_id: The escalation record ID to update.
        outcome_after_action: What happened (success, failed, aborted, skipped).

    Returns:
        True if the record was updated, False if not found.
    """
    with self._get_connection() as conn:
        cursor = conn.execute(
            """
            UPDATE escalation_decisions
            SET outcome_after_action = ?
            WHERE id = ?
            """,
            (outcome_after_action, escalation_id),
        )
        updated = cursor.rowcount > 0

    if updated:
        _logger.debug(
            f"Updated escalation {escalation_id} outcome: {outcome_after_action}"
        )

    return updated

Functions