outcomes

outcomes

Outcome recording and pattern detection for learning.

Attributes

Classes

SheetOutcome dataclass

SheetOutcome(sheet_id, job_id, validation_results, execution_duration, retry_count, completion_mode_used, final_status, validation_pass_rate, success_without_retry, patterns_detected=list(), timestamp=(lambda: now(tz=UTC))(), failure_category_counts=dict(), semantic_patterns=list(), fix_suggestions=list(), patterns_applied=list(), stdout_tail='', stderr_tail='', error_history=list(), grounding_passed=None, grounding_confidence=None, grounding_guidance=None)

Structured outcome data from a completed sheet execution.

This dataclass captures all relevant information about a sheet execution for learning and pattern detection purposes.

Attributes

failure_category_counts class-attribute instance-attribute

failure_category_counts = field(default_factory=dict)

Counts of each failure category from validation results.

Example: {'missing': 2, 'stale': 1, 'malformed': 0} Categories: missing, malformed, incomplete, stale, error

semantic_patterns class-attribute instance-attribute

semantic_patterns = field(default_factory=list)

Extracted semantic patterns from failure_reason fields.

Example: ['file not created', 'pattern not found', 'content empty'] These are normalized phrases that can be aggregated across outcomes.

fix_suggestions class-attribute instance-attribute

fix_suggestions = field(default_factory=list)

Collected suggested_fix values from failed validations.

Example: ['Ensure file is created in workspace/', 'Add missing import']

patterns_applied class-attribute instance-attribute

patterns_applied = field(default_factory=list)

Pattern descriptions that were applied/injected for this sheet execution.

These are the patterns from get_relevant_patterns() that were included in the prompt. Used for effectiveness tracking: correlate patterns_applied with success_without_retry to measure which patterns actually help.

Example: ['⚠️ Common issue: file_exists validation tends to fail (seen 3x)']

stdout_tail class-attribute instance-attribute

stdout_tail = ''

Captured tail of stdout from execution.

Stores the last N characters of stdout for pattern extraction. Used by OutputPatternExtractor to detect error patterns.

stderr_tail class-attribute instance-attribute

stderr_tail = ''

Captured tail of stderr from execution.

Stores the last N characters of stderr for pattern extraction. Used by OutputPatternExtractor to detect error patterns.

error_history class-attribute instance-attribute

error_history = field(default_factory=list)

History of errors encountered during execution.

Each entry is a dict with error_code, category, message, etc. Used by _detect_error_code_patterns() for recurring error analysis.

Example: [{'error_code': 'E009', 'category': 'transient', 'message': 'Rate limited'}]

grounding_passed class-attribute instance-attribute

grounding_passed = None

Whether all grounding hooks passed (None if grounding not enabled).

Used to correlate grounding outcomes with validation results and pattern effectiveness over time.

grounding_confidence class-attribute instance-attribute

grounding_confidence = None

Average confidence across grounding hooks (0.0-1.0, None if not enabled).

Higher confidence means external validators had high certainty in their results. Can be correlated with success_without_retry to identify reliable grounding sources.

grounding_guidance class-attribute instance-attribute

grounding_guidance = None

Recovery guidance from failed grounding hooks (None if passed or not enabled).

Captures actionable suggestions from external validators that failed, useful for pattern detection and learning what recovery strategies work.

OutcomeStore

Bases: Protocol

Protocol for outcome storage backends.

Provides async methods for recording outcomes and querying for similar past outcomes to inform execution decisions.

Functions

record async

record(outcome)

Record a sheet outcome for future learning.

Source code in src/marianne/learning/outcomes.py

async def record(self, outcome: SheetOutcome) -> None:
    """Record a sheet outcome for future learning."""
    ...

query_similar async

query_similar(context, limit=10)

Query for similar past outcomes based on context.

Source code in src/marianne/learning/outcomes.py

async def query_similar(
    self, context: dict[str, Any], limit: int = 10
) -> list[SheetOutcome]:
    """Query for similar past outcomes based on context."""
    ...

get_patterns async

get_patterns(job_name)

Get detected patterns for a specific job.

Source code in src/marianne/learning/outcomes.py

async def get_patterns(self, job_name: str) -> list[str]:
    """Get detected patterns for a specific job."""
    ...

get_relevant_patterns async

get_relevant_patterns(context, limit=5)

Get pattern descriptions relevant to the given context.

Source code in src/marianne/learning/outcomes.py

async def get_relevant_patterns(
    self, context: dict[str, Any], limit: int = 5
) -> list[str]:
    """Get pattern descriptions relevant to the given context."""
    ...

JsonOutcomeStore

JsonOutcomeStore(store_path)

JSON-file based outcome store implementation.

Stores outcomes in a JSON file with atomic saves, following the same pattern as JsonStateBackend.

Initialize the JSON outcome store.

Parameters:

Name	Type	Description	Default
store_path	Path	Path to the JSON file for storing outcomes.	required

Source code in src/marianne/learning/outcomes.py

def __init__(self, store_path: Path) -> None:
    """Initialize the JSON outcome store.

    Args:
        store_path: Path to the JSON file for storing outcomes.
    """
    self.store_path = store_path
    self._outcomes: list[SheetOutcome] = []
    self._loaded: bool = False
    self._cached_patterns: list[DetectedPattern] | None = None
    self._patterns_outcome_count: int = 0

Functions

record async

record(outcome)

Record a sheet outcome to the store.

After recording, if there are enough outcomes (>= 5), pattern detection is run and patterns_detected is populated on the outcome.

Parameters:

Name	Type	Description	Default
outcome	SheetOutcome	The sheet outcome to record.	required

Source code in src/marianne/learning/outcomes.py

async def record(self, outcome: SheetOutcome) -> None:
    """Record a sheet outcome to the store.

    After recording, if there are enough outcomes (>= 5), pattern
    detection is run and patterns_detected is populated on the outcome.

    Args:
        outcome: The sheet outcome to record.
    """
    # Load existing outcomes first to avoid overwriting on fresh store instance
    await self._load()

    # Check if this sheet already has an outcome (update rather than duplicate)
    existing_idx = next(
        (
            i for i, existing in enumerate(self._outcomes)
            if existing.sheet_id == outcome.sheet_id
        ),
        None,
    )
    if existing_idx is not None:
        self._outcomes[existing_idx] = outcome
    else:
        self._outcomes.append(outcome)

    # Invalidate cached patterns when outcomes change
    self._cached_patterns = None

    # Detect patterns after accumulating enough data
    if len(self._outcomes) >= 5:
        patterns = await self.detect_patterns()
        # Populate patterns_detected with top pattern descriptions
        outcome.patterns_detected = [
            p.to_prompt_guidance() for p in patterns[:3]
        ]

    await self._save()

query_similar async

query_similar(context, limit=10)

Query for similar past outcomes.

Currently returns recent outcomes for the same job_id. Future: implement semantic similarity matching.

Parameters:

Name	Type	Description	Default
context	dict[str, Any]	Context dict containing job_id and other metadata.	required
limit	int	Maximum number of outcomes to return.	10

Returns:

Type	Description
list[SheetOutcome]	List of similar sheet outcomes.

Source code in src/marianne/learning/outcomes.py

async def query_similar(
    self, context: dict[str, Any], limit: int = 10
) -> list[SheetOutcome]:
    """Query for similar past outcomes.

    Currently returns recent outcomes for the same job_id.
    Future: implement semantic similarity matching.

    Args:
        context: Context dict containing job_id and other metadata.
        limit: Maximum number of outcomes to return.

    Returns:
        List of similar sheet outcomes.
    """
    await self._load()
    job_id = context.get("job_id")
    if not job_id:
        return self._outcomes[-limit:]

    matching = [outcome for outcome in self._outcomes if outcome.job_id == job_id]
    return matching[-limit:]

get_patterns async

get_patterns(job_name)

Get detected patterns for a job.

Parameters:

Name	Type	Description	Default
job_name	str	The job name to get patterns for.	required

Returns:

Type	Description
list[str]	List of pattern strings detected across outcomes.

Source code in src/marianne/learning/outcomes.py

async def get_patterns(self, job_name: str) -> list[str]:
    """Get detected patterns for a job.

    Args:
        job_name: The job name to get patterns for.

    Returns:
        List of pattern strings detected across outcomes.
    """
    await self._load()
    patterns: set[str] = set()
    for outcome in self._outcomes:
        if outcome.job_id == job_name:
            patterns.update(outcome.patterns_detected)
    return list(patterns)

detect_patterns async

detect_patterns()

Detect patterns from all recorded outcomes.

Uses PatternDetector to analyze historical outcomes and identify recurring patterns that can inform future executions. Results are cached and invalidated when outcomes change (record/load).

Returns:

Type	Description
list[DetectedPattern]	List of DetectedPattern objects sorted by confidence.

Source code in src/marianne/learning/outcomes.py

async def detect_patterns(self) -> list["DetectedPattern"]:
    """Detect patterns from all recorded outcomes.

    Uses PatternDetector to analyze historical outcomes and
    identify recurring patterns that can inform future executions.
    Results are cached and invalidated when outcomes change (record/load).

    Returns:
        List of DetectedPattern objects sorted by confidence.
    """
    from marianne.learning.patterns import PatternDetector

    await self._load()
    if not self._outcomes:
        return []

    # Return cached patterns if outcome list hasn't changed
    if (
        self._cached_patterns is not None
        and self._patterns_outcome_count == len(self._outcomes)
    ):
        return self._cached_patterns

    detector = PatternDetector(self._outcomes)
    self._cached_patterns = detector.detect_all()
    self._patterns_outcome_count = len(self._outcomes)
    return self._cached_patterns

get_relevant_patterns async

get_relevant_patterns(context, limit=5)

Get pattern descriptions relevant to the given context.

This method detects patterns, matches them to the context, and returns human-readable descriptions suitable for prompt injection.

Parameters:

Name	Type	Description	Default
context	dict[str, Any]	Context dict containing job_id, sheet_num, validation_types, etc.	required
limit	int	Maximum number of patterns to return.	5

Returns:

Type	Description
list[str]	List of pattern description strings for prompt injection.

Source code in src/marianne/learning/outcomes.py

async def get_relevant_patterns(
    self,
    context: dict[str, Any],
    limit: int = 5,
) -> list[str]:
    """Get pattern descriptions relevant to the given context.

    This method detects patterns, matches them to the context,
    and returns human-readable descriptions suitable for prompt injection.

    Args:
        context: Context dict containing job_id, sheet_num, validation_types, etc.
        limit: Maximum number of patterns to return.

    Returns:
        List of pattern description strings for prompt injection.
    """
    from marianne.learning.patterns import PatternApplicator, PatternMatcher

    # Detect all patterns
    all_patterns = await self.detect_patterns()
    if not all_patterns:
        return []

    # Match patterns to context
    matcher = PatternMatcher(all_patterns)
    matched = matcher.match(context, limit=limit)

    # Convert to prompt-ready descriptions
    applicator = PatternApplicator(matched)
    return applicator.get_pattern_descriptions()