Reporting

Matplotlib-based visualisation for coherence analysis and phase dynamics. Requires the plot optional extra:

pip install scpn-phase-orchestrator[plot]

Purpose in operations

This is the production evidence surface for post-run understanding. CoherencePlot intentionally reads from the same audit records that operators and reviewers already use for compliance and replay, so every figure can be traced back to immutable records.

Use this API when:

• you need trend evidence for safety gate review (coherence and regime trajectories),
• you need a compact action audit before a change-control meeting,
• or you need explainable outputs from a run that is already fully replayable.

The module is deterministic by construction: identical audit inputs produce identical plot assets and JSON summaries.

CoherencePlot

The reporting module provides a single class CoherencePlot that consumes JSONL audit log data and produces diagnostic figures.

CoherencePlot(log_data: list[dict])

The constructor accepts a list of parsed audit log records (from ReplayEngine.load() or direct JSON parsing). It filters to step records containing "step" and "layers" fields.

Available plots

Method	Output	Description
plot_r_timeline(output_path)	PNG	Per-layer R over simulation steps
plot_regime_timeline(output_path)	PNG	Regime epochs as coloured horizontal bands
plot_action_audit(output_path)	PNG	R(t) with actuation event markers
plot_amplitude_timeline(output_path)	PNG	Mean amplitude and subcritical fraction
plot_pac_heatmap(output_path)	PNG	Phase-amplitude coupling matrix

All methods return Path to the saved figure.

Regime colour conventions

Regime	Colour	Hex
NOMINAL	Green	#2ecc71
DEGRADED	Orange	#f39c12
CRITICAL	Red	#e74c3c
RECOVERY	Blue	#3498db

Pipeline integration

AuditLogger.log_step() ──→ audit.jsonl
                                │
                                ↓
                       ReplayEngine.load()
                                │
                                ↓
                       CoherencePlot(log_data)
                                │
               ┌────────────────┼────────────────┐
               ↓                ↓                ↓
        plot_r_timeline  plot_regime_timeline  plot_action_audit
               │                │                │
               ↓                ↓                ↓
          r_timeline.png  regime.png        actions.png

The reporting module consumes audit log output. It does not connect to the engine directly — all data passes through the audit trail, ensuring that plots match the auditable record.

When the audit trail contains a run header, spo report --json-out includes the resolved binding_summary. N-channel runs also expose channel_algebra at the top level of the JSON report so downstream tools can read channel groups, derived channels, runtime evidence, and missing required channel evidence without re-parsing the binding spec.

The text report also prints a compact channel-algebra line when the audit header contains one, including required/optional/derived/delayed/uncertain counts and any missing required channel evidence.

If the audit stream includes passive integrated-information monitor records with monitor: integrated_information, the JSON summary includes an integrated_information block with latest Phi proxy values, normalised Phi values, series data, record count, and the claim boundary. The text report prints a compact line with the latest Phi proxy, normalised Phi, total integration, and number of monitor records.

Programmatic tools can use build_audit_report_summary() directly to get the same JSON-ready report payload as spo report --json-out.

Usage

from scpn_phase_orchestrator.runtime.replay import ReplayEngine
from scpn_phase_orchestrator.reporting.plots import CoherencePlot

# Load audit log
replay = ReplayEngine("audit.jsonl")
entries = replay.load()

# Generate diagnostic plots
plotter = CoherencePlot(entries)
plotter.plot_r_timeline("output/r_timeline.png")
plotter.plot_regime_timeline("output/regime.png")
plotter.plot_action_audit("output/actions.png")
plotter.plot_amplitude_timeline("output/amplitude.png")
plotter.plot_pac_heatmap("output/pac.png")

Internal extraction methods

Method	Returns	Description
_extract_r_series	(steps, n_layers, series)	Per-layer R arrays
_extract_regime_epochs	[(regime, start, end)]	Regime change boundaries
_extract_actions	(steps, r_global, knob_steps)	Action event indices
_extract_amplitude	(steps, amps, sub_frac)	Amplitude time series
_extract_pac_matrix	(n, matrix)	PAC from last log record

plots

Optional matplotlib diagnostics for audit logs.

CoherencePlot renders coherence, regime, action, amplitude, and PAC views from already-recorded audit data. Matplotlib is imported lazily and a non-interactive backend is selected for production/headless use; malformed or missing plot inputs raise ValueError rather than emitting partial figures.

Classes

CoherencePlot

CoherencePlot(log_data: list[dict[str, Any]])

Audit log visualisation from JSONL step records.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def __init__(self, log_data: list[dict[str, Any]]) -> None:
    self._data = log_data
    self._steps = [
        d for d in log_data if isinstance(d, dict) and "step" in d and "layers" in d
    ]

Methods:

plot_r_timeline

plot_r_timeline(output_path: str | Path) -> Path

Line chart of per-layer R over simulation steps.

Parameters

output_path : str | Path Destination path for the artefact.

Returns

Path Line chart of per-layer R over simulation steps.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def plot_r_timeline(self, output_path: str | Path) -> Path:
    """Line chart of per-layer R over simulation steps.

    Parameters
    ----------
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Line chart of per-layer R over simulation steps.
    """
    x, n_layers, series = self._extract_r_series()
    plt, _ = _require_matplotlib()

    fig, ax = plt.subplots(figsize=(10, 4))
    for i in range(n_layers):
        ax.plot(x, series[i], linewidth=0.8, label=f"L{i}")
    ax.set_xlabel("Step")
    ax.set_ylabel("R (order parameter)")
    ax.set_ylim(-0.05, 1.05)
    ax.legend(fontsize=6, ncol=min(n_layers, 8), loc="upper right")
    fig.tight_layout()

    out = Path(output_path)
    fig.savefig(out, dpi=150)
    plt.close(fig)
    return out

plot_regime_timeline

plot_regime_timeline(output_path: str | Path) -> Path

Coloured horizontal bands per regime epoch.

Parameters

output_path : str | Path Destination path for the artefact.

Returns

Path Coloured horizontal bands per regime epoch.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def plot_regime_timeline(self, output_path: str | Path) -> Path:
    """Coloured horizontal bands per regime epoch.

    Parameters
    ----------
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Coloured horizontal bands per regime epoch.
    """
    epochs = self._extract_regime_epochs()
    plt, Rectangle = _require_matplotlib()
    steps = self._steps

    fig, ax = plt.subplots(figsize=(10, 1.5))
    for regime, start, end in epochs:
        color = _REGIME_COLORS.get(regime, "#95a5a6")
        ax.add_patch(Rectangle((start, 0), end - start, 1, color=color, alpha=0.7))

    ax.set_xlim(steps[0]["step"], steps[-1]["step"] + 1)
    ax.set_ylim(0, 1)
    ax.set_yticks([])
    ax.set_xlabel("Step")

    # Legend
    for label, color in _REGIME_COLORS.items():
        ax.plot([], [], "s", color=color, label=label)
    ax.legend(fontsize=7, ncol=4, loc="upper center", bbox_to_anchor=(0.5, 1.35))
    fig.tight_layout()

    out = Path(output_path)
    fig.savefig(out, dpi=150, bbox_inches="tight")
    plt.close(fig)
    return out

plot_action_audit

plot_action_audit(output_path: str | Path) -> Path

Vertical markers at steps where control actions fired.

Parameters

output_path : str | Path Destination path for the artefact.

Returns

Path Vertical markers at steps where control actions fired.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def plot_action_audit(self, output_path: str | Path) -> Path:
    """Vertical markers at steps where control actions fired.

    Parameters
    ----------
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Vertical markers at steps where control actions fired.
    """
    x_all, r_global, knob_steps = self._extract_actions()
    plt, _ = _require_matplotlib()

    fig, ax = plt.subplots(figsize=(10, 4))
    ax.plot(x_all, r_global, color="#2c3e50", linewidth=0.8, label="R_global")

    palette = ["#e74c3c", "#3498db", "#2ecc71", "#9b59b6", "#f39c12"]
    knob_colors: dict[str, str] = {}
    for idx, knob in enumerate(knob_steps):
        knob_colors[knob] = palette[idx % len(palette)]
        for step_x in knob_steps[knob]:
            ax.axvline(step_x, color=knob_colors[knob], alpha=0.4, linewidth=0.6)

    ax.set_xlabel("Step")
    ax.set_ylabel("R_global")
    ax.set_ylim(-0.05, 1.05)

    for knob, color in knob_colors.items():
        ax.plot([], [], color=color, linewidth=2, label=f"action: {knob}")
    ax.legend(fontsize=7, loc="upper right")
    fig.tight_layout()

    out = Path(output_path)
    fig.savefig(out, dpi=150)
    plt.close(fig)
    return out

plot_amplitude_timeline

plot_amplitude_timeline(output_path: str | Path) -> Path

Mean amplitude per step with subcritical threshold line.

Reads 'mean_amplitude' from step records. Falls back to zero if the field is absent (phase-only simulation).

Parameters

output_path : str | Path Destination path for the artefact.

Returns

Path Mean amplitude per step with subcritical threshold line.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def plot_amplitude_timeline(self, output_path: str | Path) -> Path:
    """Mean amplitude per step with subcritical threshold line.

    Reads 'mean_amplitude' from step records. Falls back to zero
    if the field is absent (phase-only simulation).

    Parameters
    ----------
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Mean amplitude per step with subcritical threshold line.
    """
    x, amps, sub_frac = self._extract_amplitude()
    plt, _ = _require_matplotlib()

    fig, ax1 = plt.subplots(figsize=(10, 4))
    ax1.plot(x, amps, color="#2980b9", linewidth=1.0, label="mean amplitude")
    ax1.set_xlabel("Step")
    ax1.set_ylabel("Mean amplitude", color="#2980b9")
    ax1.tick_params(axis="y", labelcolor="#2980b9")

    ax2 = ax1.twinx()
    ax2.plot(
        x,
        sub_frac,
        color="#e74c3c",
        linewidth=0.8,
        linestyle="--",
        label="subcritical fraction",
    )
    ax2.set_ylabel("Subcritical fraction", color="#e74c3c")
    ax2.tick_params(axis="y", labelcolor="#e74c3c")
    ax2.set_ylim(-0.05, 1.05)

    lines1, labels1 = ax1.get_legend_handles_labels()
    lines2, labels2 = ax2.get_legend_handles_labels()
    ax1.legend(lines1 + lines2, labels1 + labels2, fontsize=7, loc="upper right")
    fig.tight_layout()

    out = Path(output_path)
    fig.savefig(out, dpi=150)
    plt.close(fig)
    return out

plot_pac_heatmap

plot_pac_heatmap(output_path: str | Path) -> Path

N x N PAC modulation index heatmap.

Reads the last 'pac_matrix' event/record from log data. The matrix should be stored as a flat list with shape (N, N).

Parameters

output_path : str | Path Destination path for the artefact.

Returns

Path N x N PAC modulation index heatmap.

Source code in src/scpn_phase_orchestrator/reporting/plots.py

def plot_pac_heatmap(self, output_path: str | Path) -> Path:
    """N x N PAC modulation index heatmap.

    Reads the last 'pac_matrix' event/record from log data.
    The matrix should be stored as a flat list with shape (N, N).

    Parameters
    ----------
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        N x N PAC modulation index heatmap.
    """
    n, mat = self._extract_pac_matrix()
    plt, _ = _require_matplotlib()

    fig, ax = plt.subplots(figsize=(6, 5))
    im = ax.imshow(mat, cmap="viridis", aspect="equal", vmin=0.0)
    ax.set_xlabel("Amplitude oscillator")
    ax.set_ylabel("Phase oscillator")
    fig.colorbar(im, ax=ax, label="Modulation index")
    fig.tight_layout()

    out = Path(output_path)
    fig.savefig(out, dpi=150)
    plt.close(fig)
    return out

Summary Builder

summary

Reusable audit report summaries for CLI, notebooks, and tools.

Functions:

build_audit_report_summary

build_audit_report_summary(
    entries: list[dict[str, object]],
    *,
    hash_chain_ok: bool,
    hash_chain_verified: int,
) -> dict[str, object]

Build a JSON-ready report summary from audit log entries.

Parameters

entries : list[dict[str, object]] Audit-log entries. hash_chain_ok : bool Whether the hash chain verified. hash_chain_verified : int Whether the hash chain verified.

Returns

dict[str, object] A JSON-ready report summary from audit log entries.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/reporting/summary.py

def build_audit_report_summary(
    entries: list[dict[str, object]],
    *,
    hash_chain_ok: bool,
    hash_chain_verified: int,
) -> dict[str, object]:
    """Build a JSON-ready report summary from audit log entries.

    Parameters
    ----------
    entries : list[dict[str, object]]
        Audit-log entries.
    hash_chain_ok : bool
        Whether the hash chain verified.
    hash_chain_verified : int
        Whether the hash chain verified.

    Returns
    -------
    dict[str, object]
        A JSON-ready report summary from audit log entries.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    records = _record_entries(entries)
    steps = [entry for entry in records if "step" in entry and "layers" in entry]
    events = [entry for entry in records if "event" in entry]
    header = _load_header(records)
    if not steps:
        raise ValueError("audit report requires at least one step record")

    n_steps = len(steps)
    n_layers = max(len(_layers(step)) for step in steps)
    r_series = [
        [_layer_r(_layers(step)[index]) for step in steps if index < len(_layers(step))]
        for index in range(n_layers)
    ]
    regime_counts: dict[str, int] = {}
    action_counts: dict[str, int] = {}
    for step in steps:
        regime = str(step.get("regime", "NOMINAL"))
        regime_counts[regime] = regime_counts.get(regime, 0) + 1
        for action in _actions(step):
            knob = str(action.get("knob", "?"))
            action_counts[knob] = action_counts.get(knob, 0) + 1

    summary: dict[str, object] = {
        "steps": n_steps,
        "layers": n_layers,
        "amplitude_mode": bool(header and header.get("amplitude_mode")),
        "final_regime": steps[-1].get("regime", "unknown"),
        "final_stability": _numeric_value(steps[-1], "stability"),
        "layer_r_mean": [
            round(sum(series) / len(series), 4) if series else 0.0
            for series in r_series
        ],
        "layer_r_final": [
            round(series[-1], 4) if series else 0.0 for series in r_series
        ],
        "regime_counts": regime_counts,
        "action_counts": action_counts,
        "events": len(events),
        "hash_chain_ok": hash_chain_ok,
        "hash_chain_verified": hash_chain_verified,
    }
    integrated_information = _integrated_information_summary(records)
    if integrated_information is not None:
        summary["integrated_information"] = integrated_information
    if header is not None:
        binding_summary = header.get("binding_summary") or header.get("binding_config")
        if isinstance(binding_summary, dict):
            summary["binding_summary"] = binding_summary
            channel_algebra = binding_summary.get("channel_algebra")
            if isinstance(channel_algebra, dict):
                summary["channel_algebra"] = channel_algebra
    return summary

Explainability

Human-readable report helpers that translate audit and supervisor records into plain diagnostic summaries for notebooks, demos, and operator-facing reports.

explainability

Human-readable explanations derived from audit records.

Explainability reports summarise regimes, transitions, layer coherence, stability metrics, action justifications, events, and hash-chain integrity. Inputs are parsed audit entries; missing step records or invalid report parameters fail explicitly instead of producing misleading empty reports.

Classes

ActionExplanation dataclass

ActionExplanation(
    step: int,
    regime: str,
    knob: str,
    scope: str,
    value: float,
    ttl_s: float,
    reason: str,
    evidence: tuple[str, ...],
)

Human-readable explanation for one control action.

ExplainabilityReport dataclass

ExplainabilityReport(
    steps: int,
    layers: int,
    hash_chain_ok: bool,
    hash_chain_verified: int,
    final_regime: str,
    final_stability: float,
    regime_counts: dict[str, int],
    regime_transitions: tuple[str, ...],
    action_explanations: tuple[ActionExplanation, ...],
    events: tuple[str, ...],
    metric_summary: tuple[str, ...],
)

Structured explainability summary derived from an audit JSONL file.

Functions:

build_explainability_report

build_explainability_report(
    entries: list[dict[str, Any]], *, max_actions: int = 12
) -> ExplainabilityReport

Build a structured explanation from parsed audit records.

Parameters

entries : list[dict[str, Any]] Audit-log entries. max_actions : int Maximum number of actions to include.

Returns

ExplainabilityReport A structured explanation from parsed audit records.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/reporting/explainability.py

def build_explainability_report(
    entries: list[dict[str, Any]],
    *,
    max_actions: int = 12,
) -> ExplainabilityReport:
    """Build a structured explanation from parsed audit records.

    Parameters
    ----------
    entries : list[dict[str, Any]]
        Audit-log entries.
    max_actions : int
        Maximum number of actions to include.

    Returns
    -------
    ExplainabilityReport
        A structured explanation from parsed audit records.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    steps = _step_records(entries)
    if not steps:
        raise ValueError("no step records in audit log")
    integrity_ok, n_verified = ReplayEngine.verify_integrity(entries)
    last = steps[-1]
    return ExplainabilityReport(
        steps=len(steps),
        layers=max(len(_layers(step)) for step in steps),
        hash_chain_ok=integrity_ok,
        hash_chain_verified=n_verified,
        final_regime=str(last.get("regime", "unknown")),
        final_stability=_numeric_value(last.get("stability", 0.0)),
        regime_counts=_regime_counts(steps),
        regime_transitions=_regime_transitions(steps),
        action_explanations=_action_explanations(steps, max_actions),
        events=_event_lines(entries),
        metric_summary=_metric_summary(steps),
    )

render_markdown

render_markdown(report: ExplainabilityReport) -> str

Render a structured report as Markdown.

Parameters

report : ExplainabilityReport The report mapping.

Returns

str Render a structured report as Markdown.

Source code in src/scpn_phase_orchestrator/reporting/explainability.py

def render_markdown(report: ExplainabilityReport) -> str:
    """Render a structured report as Markdown.

    Parameters
    ----------
    report : ExplainabilityReport
        The report mapping.

    Returns
    -------
    str
        Render a structured report as Markdown.
    """
    lines = [
        "# SCPN Phase Orchestrator Explainability Report",
        "",
        "## Summary",
        f"- Steps analysed: {report.steps}",
        f"- Layers: {report.layers}",
        f"- Final regime: {report.final_regime}",
        f"- Final stability proxy: {report.final_stability:.4f}",
        "- Hash chain: "
        f"{'OK' if report.hash_chain_ok else 'FAILED'} "
        f"({report.hash_chain_verified} records verified)",
        "",
        "## Metric Evidence",
    ]
    lines.extend(f"- {item}" for item in report.metric_summary)
    lines.extend(["", "## Regime Distribution"])
    for regime, count in sorted(report.regime_counts.items()):
        pct = 100.0 * count / report.steps
        lines.append(f"- {regime}: {count} steps ({pct:.1f}%)")
    lines.extend(["", "## Regime Transitions"])
    if report.regime_transitions:
        lines.extend(f"- {transition}" for transition in report.regime_transitions)
    else:
        lines.append("- No regime transitions recorded.")
    lines.extend(["", "## Control Action Explanations"])
    if report.action_explanations:
        for explanation in report.action_explanations:
            lines.append(
                f"- Step {explanation.step}: {explanation.knob}="
                f"{explanation.value:.4f} ({explanation.scope}, "
                f"ttl={explanation.ttl_s:.1f}s) in {explanation.regime}. "
                f"Reason: {explanation.reason}. Evidence: "
                f"{'; '.join(explanation.evidence)}."
            )
    else:
        lines.append("- No control actions recorded.")
    lines.extend(["", "## Events"])
    if report.events:
        lines.extend(f"- {event}" for event in report.events)
    else:
        lines.append("- No auxiliary events recorded.")
    return "\n".join(lines) + "\n"

write_markdown

write_markdown(
    report: ExplainabilityReport, output_path: str | Path
) -> Path

Write Markdown report and return the output path.

Parameters

report : ExplainabilityReport The report mapping. output_path : str | Path Destination path for the artefact.

Returns

Path Write Markdown report and return the output path.

Source code in src/scpn_phase_orchestrator/reporting/explainability.py

def write_markdown(report: ExplainabilityReport, output_path: str | Path) -> Path:
    """Write Markdown report and return the output path.

    Parameters
    ----------
    report : ExplainabilityReport
        The report mapping.
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Write Markdown report and return the output path.
    """
    out = Path(output_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_text(render_markdown(report), encoding="utf-8")
    return out

markdown_to_pdf_bytes

markdown_to_pdf_bytes(markdown: str) -> bytes

Return Markdown text rendered as a deterministic, dependency-free PDF.

The renderer wraps the text to the page width (upper-casing heading lines) and emits a minimal single-font PDF. It contains no timestamp or other non-deterministic field, so the bytes are reproducible for a given input.

Parameters

markdown : str The Markdown (or plain text) document to render.

Returns

bytes The rendered text PDF.

Source code in src/scpn_phase_orchestrator/reporting/explainability.py

def markdown_to_pdf_bytes(markdown: str) -> bytes:
    """Return Markdown text rendered as a deterministic, dependency-free PDF.

    The renderer wraps the text to the page width (upper-casing heading lines)
    and emits a minimal single-font PDF. It contains no timestamp or other
    non-deterministic field, so the bytes are reproducible for a given input.

    Parameters
    ----------
    markdown : str
        The Markdown (or plain text) document to render.

    Returns
    -------
    bytes
        The rendered text PDF.
    """
    return _make_pdf_bytes(_wrap_pdf_lines(markdown))

write_pdf

write_pdf(
    report: ExplainabilityReport, output_path: str | Path
) -> Path

Write a dependency-free text PDF report and return the output path.

Parameters

report : ExplainabilityReport The report mapping. output_path : str | Path Destination path for the artefact.

Returns

Path Write a dependency-free text PDF report and return the output path.

Source code in src/scpn_phase_orchestrator/reporting/explainability.py

def write_pdf(report: ExplainabilityReport, output_path: str | Path) -> Path:
    """Write a dependency-free text PDF report and return the output path.

    Parameters
    ----------
    report : ExplainabilityReport
        The report mapping.
    output_path : str | Path
        Destination path for the artefact.

    Returns
    -------
    Path
        Write a dependency-free text PDF report and return the output path.
    """
    out = Path(output_path)
    out.parent.mkdir(parents=True, exist_ok=True)
    out.write_bytes(markdown_to_pdf_bytes(render_markdown(report)))
    return out

Operator copilot

reporting.operator_copilot turns an ExplainabilityReport into a grounded question-answering surface for a control-room operator. It renders the audit evidence — regime distribution, transitions, every control action with its recorded reason and evidence, the metric summary — into the prompt and instructs the language model to answer only from that evidence and to decline when it is silent, so the model explains and locates what the audit records rather than inventing control history or recommending actuation. The model is any provider with a complete(prompt) -> str method (a local HTTP model, or a deterministic stub for tests), so no network backend is required and the prompt is fully testable.

operator_copilot

A grounded LLM copilot that answers operator questions from audit evidence.

The copilot turns a hash-verified :class:~scpn_phase_orchestrator.reporting. explainability.ExplainabilityReport into a question-answering surface for a control-room operator. It does not let the language model speak freely: it renders the report — regime distribution, transitions, every control action with its recorded reason and evidence, the metric summary — into the prompt, and instructs the model to answer only from that evidence and to decline when the evidence is silent. The model therefore explains and locates what the audit already records; it does not invent control history or recommend actuation.

The language model is any provider with a complete(prompt) -> str method (:class:OperatorLLM) — a local HTTP model, or a deterministic stub for tests — so no network backend is required and the prompt construction is fully testable.

Classes

OperatorLLM

Bases: Protocol

A language-model backend that completes a prompt to an answer.

Methods:

complete

complete(prompt: str) -> str

Return the model completion for prompt.

Parameters

prompt : str The grounded operator prompt.

Returns

str The model's answer.

Source code in src/scpn_phase_orchestrator/reporting/operator_copilot.py

def complete(self, prompt: str) -> str:
    """Return the model completion for ``prompt``.

    Parameters
    ----------
    prompt : str
        The grounded operator prompt.

    Returns
    -------
    str
        The model's answer.
    """
    ...

OperatorCopilot dataclass

OperatorCopilot(
    llm: OperatorLLM,
    report: ExplainabilityReport,
    max_actions: int = 12,
)

A grounded operator copilot over one explainability report.

Parameters

llm : OperatorLLM The language-model backend. report : ExplainabilityReport The hash-verified audit evidence the answers are grounded in. max_actions : int The most control actions to include in the grounding context.

Methods:

answer

answer(question: str) -> str

Answer an operator question grounded in the audit evidence.

Parameters

question : str The operator's question.

Returns

str The grounded model answer.

Raises

ValueError If question is empty.

Source code in src/scpn_phase_orchestrator/reporting/operator_copilot.py

def answer(self, question: str) -> str:
    """Answer an operator question grounded in the audit evidence.

    Parameters
    ----------
    question : str
        The operator's question.

    Returns
    -------
    str
        The grounded model answer.

    Raises
    ------
    ValueError
        If ``question`` is empty.
    """
    if not question.strip():
        raise ValueError("question must not be empty")
    return self.llm.complete(self.build_prompt(question))

build_prompt

build_prompt(question: str) -> str

Render the grounded prompt for an operator question.

Parameters

question : str The operator's question.

Returns

str The prompt: the grounding instruction, the rendered evidence, and the question.

Source code in src/scpn_phase_orchestrator/reporting/operator_copilot.py

def build_prompt(self, question: str) -> str:
    """Render the grounded prompt for an operator question.

    Parameters
    ----------
    question : str
        The operator's question.

    Returns
    -------
    str
        The prompt: the grounding instruction, the rendered evidence, and the
        question.
    """
    lines = [_GROUNDING_INSTRUCTION, "", "AUDIT EVIDENCE:"]
    report = self.report
    chain = "verified" if report.hash_chain_ok else "NOT verified"
    lines.append(
        f"- {report.steps} steps over {report.layers} layers; "
        f"hash chain {chain} ({report.hash_chain_verified} records)."
    )
    lines.append(
        f"- Final regime {report.final_regime}; "
        f"final stability {report.final_stability:.4f}."
    )
    distribution = ", ".join(
        f"{regime}={count}"
        for regime, count in sorted(report.regime_counts.items())
    )
    lines.append(f"- Regime distribution: {distribution or 'none recorded'}.")
    if report.regime_transitions:
        lines.append(
            f"- Regime transitions: {' -> '.join(report.regime_transitions)}."
        )
    if report.action_explanations:
        lines.append("- Control actions:")
        for action in report.action_explanations[: self.max_actions]:
            evidence = "; ".join(action.evidence) if action.evidence else "none"
            lines.append(
                f"  - step {action.step} [{action.regime}]: "
                f"{action.knob}@{action.scope}={action.value:.4f} "
                f"(ttl {action.ttl_s:.1f}s) — {action.reason} "
                f"(evidence: {evidence})"
            )
    else:
        lines.append("- Control actions: none recorded.")
    if report.metric_summary:
        lines.append(f"- Metrics: {'; '.join(report.metric_summary)}.")
    if report.events:
        lines.append(f"- Events: {'; '.join(report.events)}.")
    lines.extend(["", f"OPERATOR QUESTION: {question}", "", "ANSWER:"])
    return "\n".join(lines)