Index

healing

Self-healing module for Marianne.

Provides automatic diagnosis and remediation of common configuration and execution errors. The healing system activates when: 1. All retries have been exhausted 2. The error is in a healable category 3. The --self-healing flag is enabled

Public exports: - ErrorContext: Rich diagnostic context for failed executions - DiagnosisEngine: Analyzes errors and suggests remedies - Remedy: Protocol for remediation actions - RemedyCategory: AUTOMATIC, SUGGESTED, or DIAGNOSTIC - RemedyRegistry: Registry of available remedies - SelfHealingCoordinator: Orchestrates the healing process - HealingReport: Results of a healing attempt - create_default_registry: Factory for built-in remedies

Classes

ErrorContext dataclass

ErrorContext(error_code, error_message, error_category, exception=None, exit_code=None, signal=None, stdout_tail='', stderr_tail='', config_path=None, config=None, workspace=None, sheet_number=0, working_directory=None, environment=dict(), retry_count=0, max_retries=0, previous_errors=list(), raw_config_yaml=None, validation_details=list())

Rich context gathered when an error occurs.

Provides all information needed by the diagnosis engine and remedies to understand and potentially fix the error.

Attributes:

Name	Type	Description
error_code	str	Structured error code (e.g., E601, E304)
error_message	str	Human-readable error description
error_category	str	High-level category (preflight, configuration, etc.)
exception	Exception | None	Original exception if available
exit_code	int | None	Process exit code (if applicable)
signal	int | None	Termination signal (if applicable)
stdout_tail	str	Last portion of stdout output
stderr_tail	str	Last portion of stderr output
config_path	Path | None	Path to the job configuration file
config	JobConfig | None	Parsed JobConfig object
workspace	Path | None	Workspace directory path
sheet_number	int	Current sheet number
working_directory	Path | None	Backend working directory
environment	dict[str, str]	Relevant environment variables
retry_count	int	Number of retries attempted
max_retries	int	Maximum retries configured
previous_errors	list[str]	Error codes from previous attempts

Functions

from_execution_result classmethod

from_execution_result(result, config, config_path, sheet_number, error_code, error_message, error_category, retry_count=0, max_retries=0, previous_errors=None)

Create context from an execution result.

Parameters:

Name	Type	Description	Default
result	ExecutionResult	The failed execution result.	required
config	JobConfig	Job configuration.	required
config_path	Path | None	Path to config file.	required
sheet_number	int	Current sheet number.	required
error_code	str	Classified error code.	required
error_message	str	Error message.	required
error_category	str	Error category.	required
retry_count	int	Current retry count.	0
max_retries	int	Maximum retries allowed.	0
previous_errors	list[str] | None	Error codes from previous attempts.	None

Returns:

Type	Description
ErrorContext	ErrorContext with all diagnostic information.

Source code in src/marianne/healing/context.py

@classmethod
def from_execution_result(
    cls,
    result: "ExecutionResult",
    config: "JobConfig",
    config_path: Path | None,
    sheet_number: int,
    error_code: str,
    error_message: str,
    error_category: str,
    retry_count: int = 0,
    max_retries: int = 0,
    previous_errors: list[str] | None = None,
) -> "ErrorContext":
    """Create context from an execution result.

    Args:
        result: The failed execution result.
        config: Job configuration.
        config_path: Path to config file.
        sheet_number: Current sheet number.
        error_code: Classified error code.
        error_message: Error message.
        error_category: Error category.
        retry_count: Current retry count.
        max_retries: Maximum retries allowed.
        previous_errors: Error codes from previous attempts.

    Returns:
        ErrorContext with all diagnostic information.
    """
    import os

    # Capture relevant environment variables
    env_vars = {
        "PATH": os.environ.get("PATH", ""),
        "HOME": os.environ.get("HOME", ""),
        "ANTHROPIC_API_KEY": "***" if os.environ.get("ANTHROPIC_API_KEY") else "",
    }

    return cls(
        error_code=error_code,
        error_message=error_message,
        error_category=error_category,
        exit_code=result.exit_code,
        signal=result.exit_signal,
        stdout_tail=result.stdout[-HEALING_CONTEXT_TAIL_CHARS:] if result.stdout else "",
        stderr_tail=result.stderr[-HEALING_CONTEXT_TAIL_CHARS:] if result.stderr else "",
        config_path=config_path,
        config=config,
        workspace=config.workspace,
        sheet_number=sheet_number,
        working_directory=config.backend.working_directory or config.workspace,
        environment=env_vars,
        retry_count=retry_count,
        max_retries=max_retries,
        previous_errors=previous_errors or [],
    )

from_preflight_error classmethod

from_preflight_error(config, config_path, error_code, error_message, sheet_number=0, raw_yaml=None)

Create context from a preflight check failure.

Preflight errors occur before execution starts, so there's no ExecutionResult. This is common for validation errors like missing workspace directories or invalid templates.

Parameters:

Name	Type	Description	Default
config	JobConfig	Job configuration.	required
config_path	Path | None	Path to config file.	required
error_code	str	Preflight error code.	required
error_message	str	Error description.	required
sheet_number	int	Sheet number (0 if global).	0
raw_yaml	str | None	Raw YAML content for template analysis.	None

Returns:

Type	Description
ErrorContext	ErrorContext for preflight failures.

Source code in src/marianne/healing/context.py

@classmethod
def from_preflight_error(
    cls,
    config: "JobConfig",
    config_path: Path | None,
    error_code: str,
    error_message: str,
    sheet_number: int = 0,
    raw_yaml: str | None = None,
) -> "ErrorContext":
    """Create context from a preflight check failure.

    Preflight errors occur before execution starts, so there's
    no ExecutionResult. This is common for validation errors
    like missing workspace directories or invalid templates.

    Args:
        config: Job configuration.
        config_path: Path to config file.
        error_code: Preflight error code.
        error_message: Error description.
        sheet_number: Sheet number (0 if global).
        raw_yaml: Raw YAML content for template analysis.

    Returns:
        ErrorContext for preflight failures.
    """
    import os

    env_vars = {
        "PATH": os.environ.get("PATH", ""),
        "HOME": os.environ.get("HOME", ""),
    }

    return cls(
        error_code=error_code,
        error_message=error_message,
        error_category="preflight",
        config_path=config_path,
        config=config,
        workspace=config.workspace,
        sheet_number=sheet_number,
        working_directory=config.backend.working_directory or config.workspace,
        environment=env_vars,
        raw_config_yaml=raw_yaml,
    )

get_context_summary

get_context_summary()

Get a summary of context for logging/display.

Returns:

Type	Description
dict[str, Any]	Dictionary with key context information.

Source code in src/marianne/healing/context.py

def get_context_summary(self) -> dict[str, Any]:
    """Get a summary of context for logging/display.

    Returns:
        Dictionary with key context information.
    """
    return {
        "error_code": self.error_code,
        "error_category": self.error_category,
        "sheet_number": self.sheet_number,
        "retry_count": self.retry_count,
        "exit_code": self.exit_code,
        "workspace": str(self.workspace) if self.workspace else None,
        "has_stdout": bool(self.stdout_tail),
        "has_stderr": bool(self.stderr_tail),
    }

HealingReport dataclass

HealingReport(error_context, diagnoses=list(), actions_taken=list(), actions_skipped=list(), diagnostic_outputs=list())

Report of a self-healing attempt.

Captures what was diagnosed, what actions were taken, and what issues remain for manual intervention.

Attributes

error_context instance-attribute

error_context

The original error context that triggered healing.

diagnoses class-attribute instance-attribute

diagnoses = field(default_factory=list)

All diagnoses found by the engine.

actions_taken class-attribute instance-attribute

actions_taken = field(default_factory=list)

(remedy_name, result) for each action attempted.

actions_skipped class-attribute instance-attribute

actions_skipped = field(default_factory=list)

(remedy_name, reason) for each skipped action.

diagnostic_outputs class-attribute instance-attribute

diagnostic_outputs = field(default_factory=list)

(remedy_name, diagnostic_text) for guidance-only remedies.

any_remedies_applied property

any_remedies_applied

Check if any remedies were successfully applied.

issues_remaining property

issues_remaining

Count of issues that couldn't be auto-fixed.

should_retry property

should_retry

Whether a retry should be attempted after healing.

Returns True if any automatic remedies succeeded.

Functions

format

format(verbose=False)

Generate human-readable report.

Parameters:

Name	Type	Description	Default
verbose	bool	Include full diagnostic output.	False

Returns:

Type	Description
str	Formatted healing report string.

Source code in src/marianne/healing/coordinator.py

def format(self, verbose: bool = False) -> str:
    """Generate human-readable report.

    Args:
        verbose: Include full diagnostic output.

    Returns:
        Formatted healing report string.
    """
    lines = [
        "═" * 75,
        f"SELF-HEALING REPORT: Sheet {self.error_context.sheet_number}",
        "═" * 75,
        "",
        "Error Diagnosed:",
        f"  Code: {self.error_context.error_code}",
        f"  Message: {self.error_context.error_message}",
        "",
    ]

    # Remedies applied
    lines.append("Remedies Applied:")
    if self.actions_taken:
        for name, result in self.actions_taken:
            status = "✓" if result.success else "✗"
            category = "[AUTO]" if not self._is_suggested(name) else "[SUGGESTED]"
            lines.append(f"  {status} {category} {result.action_taken}: {result.message}")
    else:
        lines.append("  (none)")
    lines.append("")

    # Remedies skipped
    if self.actions_skipped:
        lines.append("Remedies Skipped:")
        for name, reason in self.actions_skipped:
            lines.append(f"  - {name}: {reason}")
        lines.append("")

    # Diagnostic outputs
    if self.diagnostic_outputs:
        lines.append("Remaining Issues (Manual Fix Required):")
        for name, output in self.diagnostic_outputs:
            lines.append(f"  [{name}]")
            if verbose:
                for line in output.split("\n"):
                    lines.append(f"    {line}")
            else:
                # Just show first line
                first_line = output.split("\n")[0]
                lines.append(f"    {first_line}")
        lines.append("")

    # Result message
    if self.should_retry:
        result_msg = "HEALED - Retrying sheet"
    elif self.issues_remaining == 0 and not self.any_remedies_applied:
        result_msg = "NO ACTION NEEDED"
    else:
        result_msg = f"INCOMPLETE - {self.issues_remaining} issue(s) remaining"

    lines.extend([
        f"Result: {result_msg}",
        "═" * 75,
    ])

    return "\n".join(lines)

SelfHealingCoordinator

SelfHealingCoordinator(registry, auto_confirm=False, dry_run=False, disabled_remedies=None, max_healing_attempts=2)

Orchestrates the self-healing process.

Takes an error context and coordinates: 1. Finding applicable remedies 2. Applying automatic remedies 3. Prompting for suggested remedies 4. Collecting diagnostic output

Example

registry = create_default_registry() coordinator = SelfHealingCoordinator(registry)

context = ErrorContext.from_execution_result(...) report = await coordinator.heal(context)

if report.should_retry: # Retry the sheet

Initialize the coordinator.

Parameters:

Name	Type	Description	Default
registry	RemedyRegistry	Registry of available remedies.	required
auto_confirm	bool	Auto-approve suggested remedies (--yes flag).	False
dry_run	bool	Show what would be done without changes.	False
disabled_remedies	set[str] | None	Set of remedy names to skip.	None
max_healing_attempts	int	Maximum healing cycles before giving up.	2

Source code in src/marianne/healing/coordinator.py

def __init__(
    self,
    registry: RemedyRegistry,
    auto_confirm: bool = False,
    dry_run: bool = False,
    disabled_remedies: set[str] | None = None,
    max_healing_attempts: int = 2,
) -> None:
    """Initialize the coordinator.

    Args:
        registry: Registry of available remedies.
        auto_confirm: Auto-approve suggested remedies (--yes flag).
        dry_run: Show what would be done without changes.
        disabled_remedies: Set of remedy names to skip.
        max_healing_attempts: Maximum healing cycles before giving up.
    """
    self.registry = registry
    self.auto_confirm = auto_confirm
    self.dry_run = dry_run
    self.disabled_remedies = disabled_remedies or set()
    self.max_healing_attempts = max_healing_attempts
    self._healing_attempt = 0
    self._diagnosis_engine = DiagnosisEngine(registry)

Functions

heal async

heal(context)

Run the self-healing process.

1. Find applicable remedies
2. Apply automatic remedies
3. Prompt for suggested remedies (unless auto_confirm)
4. Display diagnostic output for manual-fix issues
5. Return report of what was done

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context with diagnostic information.	required

Returns:

Type	Description
HealingReport	HealingReport with actions taken and results.

Source code in src/marianne/healing/coordinator.py

async def heal(self, context: "ErrorContext") -> HealingReport:
    """Run the self-healing process.

    1. Find applicable remedies
    2. Apply automatic remedies
    3. Prompt for suggested remedies (unless auto_confirm)
    4. Display diagnostic output for manual-fix issues
    5. Return report of what was done

    Args:
        context: Error context with diagnostic information.

    Returns:
        HealingReport with actions taken and results.
    """
    self._healing_attempt += 1

    # Check max healing attempts
    if self._healing_attempt > self.max_healing_attempts:
        return HealingReport(
            error_context=context,
            diagnoses=[],
            actions_taken=[],
            actions_skipped=[("all", "Max healing attempts exceeded")],
            diagnostic_outputs=[],
        )

    report = HealingReport(
        error_context=context,
        diagnoses=[],
        actions_taken=[],
        actions_skipped=[],
        diagnostic_outputs=[],
    )

    # Find applicable remedies
    applicable = self.registry.find_applicable(context)

    # Surface any remedy diagnosis crashes so the report doesn't mislead
    for remedy_name, error_msg in getattr(self.registry, "diagnosis_errors", []):
        report.actions_skipped.append(
            (remedy_name, f"Diagnosis crashed: {error_msg}")
        )

    for remedy, diagnosis in applicable:
        report.diagnoses.append(diagnosis)

        # Skip disabled remedies
        if remedy.name in self.disabled_remedies:
            report.actions_skipped.append(
                (remedy.name, "Disabled in configuration")
            )
            continue

        if remedy.category == RemedyCategory.AUTOMATIC:
            # Apply automatic remedies without prompting
            if self.dry_run:
                preview = remedy.preview(context)
                report.actions_skipped.append(
                    (remedy.name, f"Dry run: would {preview}")
                )
            else:
                # remedy.apply() is synchronous — safe for fast I/O ops
                result = remedy.apply(context)
                if not result.success:
                    _logger.warning(
                        "healing.remedy_failed",
                        remedy=remedy.name,
                        message=result.message,
                    )
                report.actions_taken.append((remedy.name, result))

        elif remedy.category == RemedyCategory.SUGGESTED:
            # Prompt for suggested remedies (or auto-confirm)
            if self.auto_confirm or self._prompt_user(remedy, diagnosis):
                if self.dry_run:
                    preview = remedy.preview(context)
                    report.actions_skipped.append(
                        (remedy.name, f"Dry run: would {preview}")
                    )
                else:
                    result = remedy.apply(context)
                    if not result.success:
                        _logger.warning(
                            "healing.remedy_failed",
                            remedy=remedy.name,
                            message=result.message,
                        )
                    report.actions_taken.append((remedy.name, result))
            else:
                report.actions_skipped.append(
                    (remedy.name, "User declined")
                )
                # Add diagnostic for declined suggestion
                diagnostic = remedy.generate_diagnostic(context)
                report.diagnostic_outputs.append((remedy.name, diagnostic))

        elif remedy.category == RemedyCategory.DIAGNOSTIC:
            # Diagnostic only - generate guidance
            diagnostic = remedy.generate_diagnostic(context)
            report.diagnostic_outputs.append((remedy.name, diagnostic))

    return report

reset

reset()

Reset healing attempt counter.

Call this before starting a new healing cycle for a different error.

Source code in src/marianne/healing/coordinator.py

def reset(self) -> None:
    """Reset healing attempt counter.

    Call this before starting a new healing cycle for a different error.
    """
    self._healing_attempt = 0

Diagnosis dataclass

Diagnosis(error_code, issue, explanation, suggestion, confidence, remedy_name=None, requires_confirmation=False, context=dict())

Result of analyzing an error.

Contains information about what went wrong and how to fix it. Multiple diagnoses may be returned for a single error, sorted by confidence (highest first).

Attributes:

Name	Type	Description
error_code	str	The error code being diagnosed
issue	str	What went wrong (human-readable summary)
explanation	str	Why the error happened (detailed)
suggestion	str	How to fix it (actionable)
confidence	float	0.0-1.0, higher = more certain this diagnosis is correct
remedy_name	str | None	Name of the remedy that can fix this (if any)
requires_confirmation	bool	True for suggested remedies
context	dict[str, Any]	Extra data for the remedy to use

Functions

format_short

format_short()

Format as a single-line summary.

Source code in src/marianne/healing/diagnosis.py

def format_short(self) -> str:
    """Format as a single-line summary."""
    return f"[{self.error_code}] {self.issue} ({self.confidence:.0%} confidence)"

format_full

format_full()

Format with full details.

Source code in src/marianne/healing/diagnosis.py

def format_full(self) -> str:
    """Format with full details."""
    lines = [
        f"Error: {self.error_code}",
        f"Issue: {self.issue}",
        f"Explanation: {self.explanation}",
        f"Suggestion: {self.suggestion}",
        f"Confidence: {self.confidence:.0%}",
    ]
    if self.remedy_name:
        lines.append(f"Remedy: {self.remedy_name}")
    return "\n".join(lines)

DiagnosisEngine

DiagnosisEngine(remedy_registry)

Diagnoses errors and suggests remedies.

Takes an ErrorContext and queries all registered remedies to find applicable diagnoses, returning them sorted by confidence (highest first).

Example

registry = create_default_registry() engine = DiagnosisEngine(registry)

context = ErrorContext.from_preflight_error(...) diagnoses = engine.diagnose(context)

for diagnosis in diagnoses: print(f"{diagnosis.issue}: {diagnosis.suggestion}")

Initialize the diagnosis engine.

Parameters:

Name	Type	Description	Default
remedy_registry	RemedyRegistry	Registry containing available remedies.	required

Source code in src/marianne/healing/diagnosis.py

def __init__(self, remedy_registry: "RemedyRegistry") -> None:
    """Initialize the diagnosis engine.

    Args:
        remedy_registry: Registry containing available remedies.
    """
    self.registry = remedy_registry

Functions

diagnose

diagnose(context)

Analyze error and return possible diagnoses.

Queries all registered remedies and collects their diagnoses. Results are sorted by confidence (highest first).

If a remedy's diagnose() raises an exception, a zero-confidence sentinel Diagnosis is appended with remedy_name set to the failing remedy's class name and issue describing the failure. This ensures callers can detect partial diagnosis results without a return-type change.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context with diagnostic information.	required

Returns:

Type	Description
list[Diagnosis]	List of Diagnosis objects, sorted by confidence descending.
list[Diagnosis]	Empty list if no remedies apply.

Source code in src/marianne/healing/diagnosis.py

def diagnose(self, context: "ErrorContext") -> list[Diagnosis]:
    """Analyze error and return possible diagnoses.

    Queries all registered remedies and collects their diagnoses.
    Results are sorted by confidence (highest first).

    If a remedy's ``diagnose()`` raises an exception, a zero-confidence
    sentinel Diagnosis is appended with ``remedy_name`` set to the failing
    remedy's class name and ``issue`` describing the failure. This ensures
    callers can detect partial diagnosis results without a return-type change.

    Args:
        context: Error context with diagnostic information.

    Returns:
        List of Diagnosis objects, sorted by confidence descending.
        Empty list if no remedies apply.
    """
    diagnoses: list[Diagnosis] = []

    for remedy in self.registry.all_remedies():
        try:
            diagnosis = remedy.diagnose(context)
            if diagnosis is not None:
                diagnoses.append(diagnosis)
        except Exception as e:
            remedy_name = type(remedy).__name__
            # Individual remedy failures shouldn't block diagnosis
            _logger.warning(
                "remedy_diagnosis_failed",
                remedy=remedy_name,
                error=str(e),
            )
            # Record the failure so callers know diagnosis is partial
            diagnoses.append(Diagnosis(
                error_code=context.error_code,
                issue=f"Remedy {remedy_name} failed during diagnosis: {e}",
                explanation="This remedy could not be evaluated due to an internal error.",
                suggestion="Check logs for details; the remedy may need a bug fix.",
                confidence=0.0,
                remedy_name=None,
                context={"failed_remedy": remedy_name, "error": str(e)},
            ))

    # Sort by confidence, highest first
    diagnoses.sort(key=lambda d: d.confidence, reverse=True)
    return diagnoses

get_primary_diagnosis

get_primary_diagnosis(context)

Get the highest-confidence diagnosis.

Convenience method when you only care about the best match.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context.	required

Returns:

Type	Description
Diagnosis | None	Highest-confidence Diagnosis, or None if no remedies apply.

Source code in src/marianne/healing/diagnosis.py

def get_primary_diagnosis(self, context: "ErrorContext") -> Diagnosis | None:
    """Get the highest-confidence diagnosis.

    Convenience method when you only care about the best match.

    Args:
        context: Error context.

    Returns:
        Highest-confidence Diagnosis, or None if no remedies apply.
    """
    diagnoses = self.diagnose(context)
    return diagnoses[0] if diagnoses else None

get_automatic_diagnoses

get_automatic_diagnoses(context)

Get diagnoses for automatic remedies only.

Filters to only diagnoses that can be auto-applied without user confirmation.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context.	required

Returns:

Type	Description
list[Diagnosis]	List of diagnoses from AUTOMATIC remedies.

Source code in src/marianne/healing/diagnosis.py

def get_automatic_diagnoses(self, context: "ErrorContext") -> list[Diagnosis]:
    """Get diagnoses for automatic remedies only.

    Filters to only diagnoses that can be auto-applied without
    user confirmation.

    Args:
        context: Error context.

    Returns:
        List of diagnoses from AUTOMATIC remedies.
    """
    from marianne.healing.remedies.base import RemedyCategory

    diagnoses = self.diagnose(context)
    return [
        d
        for d in diagnoses
        if d.remedy_name and not d.requires_confirmation
        and self._get_remedy_category(d.remedy_name) == RemedyCategory.AUTOMATIC
    ]

RemedyRegistry

RemedyRegistry()

Registry of available remedies.

Maintains a list of remedy instances and provides methods to query them by name or find applicable remedies for a given error context.

Example

registry = RemedyRegistry() registry.register(CreateMissingWorkspaceRemedy())

Find by name

remedy = registry.get_by_name("create_missing_workspace")

Find all that apply to an error

applicable = registry.find_applicable(context)

Initialize an empty registry.

Source code in src/marianne/healing/registry.py

def __init__(self) -> None:
    """Initialize an empty registry."""
    self._remedies: list[Remedy] = []

Functions

register

register(remedy)

Register a remedy.

Parameters:

Name	Type	Description	Default
remedy	Remedy	Remedy instance to register.	required

Source code in src/marianne/healing/registry.py

def register(self, remedy: "Remedy") -> None:
    """Register a remedy.

    Args:
        remedy: Remedy instance to register.
    """
    self._remedies.append(remedy)

all_remedies

all_remedies()

Get all registered remedies.

Returns:

Type	Description
list[Remedy]	List of all registered remedy instances.

Source code in src/marianne/healing/registry.py

def all_remedies(self) -> list["Remedy"]:
    """Get all registered remedies.

    Returns:
        List of all registered remedy instances.
    """
    return list(self._remedies)

get_by_name

get_by_name(name)

Get remedy by name.

Parameters:

Name	Type	Description	Default
name	str	Unique remedy identifier.	required

Returns:

Type	Description
Remedy | None	Remedy if found, None otherwise.

Source code in src/marianne/healing/registry.py

def get_by_name(self, name: str) -> "Remedy | None":
    """Get remedy by name.

    Args:
        name: Unique remedy identifier.

    Returns:
        Remedy if found, None otherwise.
    """
    for remedy in self._remedies:
        if remedy.name == name:
            return remedy
    return None

find_applicable

find_applicable(context)

Find all remedies that apply to the error.

Queries each registered remedy and collects those that return a diagnosis. Results are sorted by diagnosis confidence (highest first).

Remedy diagnosis crashes are logged and recorded in self.diagnosis_errors so callers can report them instead of showing misleading "NO ACTION NEEDED" messages.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context with diagnostic information.	required

Returns:

Type	Description
list[tuple[Remedy, Diagnosis]]	List of (remedy, diagnosis) tuples sorted by confidence.

Source code in src/marianne/healing/registry.py

def find_applicable(
    self,
    context: "ErrorContext",
) -> list[tuple["Remedy", "Diagnosis"]]:
    """Find all remedies that apply to the error.

    Queries each registered remedy and collects those that
    return a diagnosis. Results are sorted by diagnosis
    confidence (highest first).

    Remedy diagnosis crashes are logged and recorded in
    ``self.diagnosis_errors`` so callers can report them
    instead of showing misleading "NO ACTION NEEDED" messages.

    Args:
        context: Error context with diagnostic information.

    Returns:
        List of (remedy, diagnosis) tuples sorted by confidence.
    """
    applicable: list[tuple[Remedy, Diagnosis]] = []
    self.diagnosis_errors: list[tuple[str, str]] = []

    for remedy in self._remedies:
        try:
            diagnosis = remedy.diagnose(context)
            if diagnosis is not None:
                applicable.append((remedy, diagnosis))
        except Exception as exc:
            # Individual remedy failures shouldn't block finding others,
            # but we must record them so the HealingReport doesn't show
            # "NO ACTION NEEDED" when a remedy actually crashed.
            _logger.warning(
                "Remedy %s.diagnose() raised exception",
                remedy.name,
                exc_info=True,
            )
            self.diagnosis_errors.append((remedy.name, str(exc)))

    # Sort by diagnosis confidence, highest first
    applicable.sort(key=lambda x: x[1].confidence, reverse=True)
    return applicable

count

count()

Get the number of registered remedies.

Returns:

Type	Description
int	Number of remedies in the registry.

Source code in src/marianne/healing/registry.py

def count(self) -> int:
    """Get the number of registered remedies.

    Returns:
        Number of remedies in the registry.
    """
    return len(self._remedies)

Remedy

Bases: Protocol

Protocol for remediation actions.

Remedies diagnose specific error patterns and can optionally apply fixes. Each remedy must implement: - name: Unique identifier - category: AUTOMATIC, SUGGESTED, or DIAGNOSTIC - risk_level: LOW, MEDIUM, or HIGH - description: Human-readable explanation - diagnose(): Check if this remedy applies - preview(): Show what would change - apply(): Make the actual changes - rollback(): Undo changes if needed - generate_diagnostic(): Provide guidance for manual fix

Attributes

name property

name

Unique identifier for this remedy.

category property

category

How this remedy should be applied.

risk_level property

risk_level

Risk level of applying this remedy.

description property

description

Human-readable description of what this remedy does.

Functions

diagnose

diagnose(context)

Check if this remedy applies to the error.

Returns Diagnosis if applicable, None otherwise. The diagnosis includes confidence score and fix suggestion.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context with diagnostic information.	required

Returns:

Type	Description
Diagnosis | None	Diagnosis if this remedy can help, None otherwise.

Source code in src/marianne/healing/remedies/base.py

def diagnose(self, context: "ErrorContext") -> "Diagnosis | None":
    """Check if this remedy applies to the error.

    Returns Diagnosis if applicable, None otherwise.
    The diagnosis includes confidence score and fix suggestion.

    Args:
        context: Error context with diagnostic information.

    Returns:
        Diagnosis if this remedy can help, None otherwise.
    """
    ...

preview

preview(context)

Show what would be changed without making changes.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context.	required

Returns:

Type	Description
str	Human-readable description of planned changes.

Source code in src/marianne/healing/remedies/base.py

def preview(self, context: "ErrorContext") -> str:
    """Show what would be changed without making changes.

    Args:
        context: Error context.

    Returns:
        Human-readable description of planned changes.
    """
    ...

apply

apply(context)

Apply the remedy.

Only called if diagnose() returned a Diagnosis and: - category == AUTOMATIC, or - category == SUGGESTED and user confirmed

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context.	required

Returns:

Type	Description
RemedyResult	Result with success status and details.

Source code in src/marianne/healing/remedies/base.py

def apply(self, context: "ErrorContext") -> RemedyResult:
    """Apply the remedy.

    Only called if diagnose() returned a Diagnosis and:
    - category == AUTOMATIC, or
    - category == SUGGESTED and user confirmed

    Args:
        context: Error context.

    Returns:
        Result with success status and details.
    """
    ...

rollback

rollback(result)

Undo the remedy if possible.

Called if remedy was applied but subsequent validation failed.

Parameters:

Name	Type	Description	Default
result	RemedyResult	The result from apply().	required

Returns:

Type	Description
bool	True if rollback succeeded, False otherwise.

Source code in src/marianne/healing/remedies/base.py

def rollback(self, result: RemedyResult) -> bool:
    """Undo the remedy if possible.

    Called if remedy was applied but subsequent validation failed.

    Args:
        result: The result from apply().

    Returns:
        True if rollback succeeded, False otherwise.
    """
    ...

generate_diagnostic

generate_diagnostic(context)

Generate detailed diagnostic message.

Called for DIAGNOSTIC category, or when user declines SUGGESTED.

Parameters:

Name	Type	Description	Default
context	ErrorContext	Error context.	required

Returns:

Type	Description
str	Formatted guidance for manual fix.

Source code in src/marianne/healing/remedies/base.py

def generate_diagnostic(self, context: "ErrorContext") -> str:
    """Generate detailed diagnostic message.

    Called for DIAGNOSTIC category, or when user declines SUGGESTED.

    Args:
        context: Error context.

    Returns:
        Formatted guidance for manual fix.
    """
    ...

RemedyCategory

Bases: str, Enum

Determines how a remedy is applied.

AUTOMATIC: Applied without asking (safe, reversible operations) SUGGESTED: Requires user confirmation (modifies files) DIAGNOSTIC: Cannot auto-fix, provides guidance only

RemedyResult dataclass

RemedyResult(success, message, action_taken, rollback_command=None, created_paths=list(), modified_files=list(), backup_paths=list())

Result of applying a remedy.

Tracks what was done and how to undo it if needed.

Attributes

success instance-attribute

success

Whether the remedy was applied successfully.

message instance-attribute

message

Human-readable description of what happened.

action_taken instance-attribute

action_taken

Brief description of the action taken.

rollback_command class-attribute instance-attribute

rollback_command = None

Shell command to undo the remedy (if possible).

created_paths class-attribute instance-attribute

created_paths = field(default_factory=list)

Paths that were created by the remedy.

modified_files class-attribute instance-attribute

modified_files = field(default_factory=list)

Files that were modified by the remedy.

backup_paths class-attribute instance-attribute

backup_paths = field(default_factory=list)

Backup files created before modification.

Functions

__str__

__str__()

Human-readable string representation.

Source code in src/marianne/healing/remedies/base.py

def __str__(self) -> str:
    """Human-readable string representation."""
    status = "✓" if self.success else "✗"
    return f"[{status}] {self.action_taken}: {self.message}"

RiskLevel

Bases: str, Enum

Risk level of applying the remedy.

Used to inform users about the potential impact of the fix.

Functions

create_default_registry

create_default_registry()

Create registry with all built-in remedies.

This is the primary factory for getting a fully-configured remedy registry. It registers:

Automatic remedies (safe, apply without confirmation): - CreateMissingWorkspaceRemedy - CreateMissingParentDirsRemedy - FixPathSeparatorsRemedy

Suggested remedies (require user confirmation): - SuggestJinjaFixRemedy

Diagnostic remedies (provide guidance only): - DiagnoseAuthErrorRemedy - DiagnoseMissingCLIRemedy

Returns:

Type	Description
RemedyRegistry	RemedyRegistry with all built-in remedies registered.

Source code in src/marianne/healing/registry.py

def create_default_registry() -> RemedyRegistry:
    """Create registry with all built-in remedies.

    This is the primary factory for getting a fully-configured
    remedy registry. It registers:

    Automatic remedies (safe, apply without confirmation):
    - CreateMissingWorkspaceRemedy
    - CreateMissingParentDirsRemedy
    - FixPathSeparatorsRemedy

    Suggested remedies (require user confirmation):
    - SuggestJinjaFixRemedy

    Diagnostic remedies (provide guidance only):
    - DiagnoseAuthErrorRemedy
    - DiagnoseMissingCLIRemedy

    Returns:
        RemedyRegistry with all built-in remedies registered.
    """
    from marianne.healing.remedies.diagnostics import (
        DiagnoseAuthErrorRemedy,
        DiagnoseMissingCLIRemedy,
    )
    from marianne.healing.remedies.jinja import SuggestJinjaFixRemedy
    from marianne.healing.remedies.paths import (
        CreateMissingParentDirsRemedy,
        CreateMissingWorkspaceRemedy,
        FixPathSeparatorsRemedy,
    )

    registry = RemedyRegistry()

    # Automatic remedies (safe, apply without asking)
    registry.register(CreateMissingWorkspaceRemedy())
    registry.register(CreateMissingParentDirsRemedy())
    registry.register(FixPathSeparatorsRemedy())

    # Suggested remedies (ask user first)
    registry.register(SuggestJinjaFixRemedy())

    # Diagnostic-only (provide guidance)
    registry.register(DiagnoseAuthErrorRemedy())
    registry.register(DiagnoseMissingCLIRemedy())

    return registry