Agent belt for #31

.github/scripts/agents_pr_meta_keepalive.js

@@ -242,7 +242,7 @@ function extractIssueNumberFromPull(pull) {
     }
     // Skip non-issue refs like "Run #123", "run #123", "attempt #2"
     const preceding = bodyText.slice(Math.max(0, match.index - 20), match.index);
-    if (/\b(?:run|attempt|step|job|check|task|version|v)\s*$/i.test(preceding)) {
+    if (/\b(?:run|attempt|step|job|check|version|v)\s*$/i.test(preceding)) {
       continue;
     }
     candidates.push(match[1]);

README.md

@@ -14,6 +14,39 @@ A programmatic replacement for the MOSERS spreadsheet workflow used to evaluate
 - A run folder with a manifest (inputs, hashes, warnings, summaries)
 - Optional “distribution” deliverables (static PPT/PDF) that do not prompt for link refresh
 
+## Output Generator Configuration
+
+Outputs are registered per run via `output_generators` in YAML config files. Each entry has:
+
+- `name`: unique identifier for the generator
+- `registration`: either `builtin:<id>` or `<python_module>:<symbol>`
+- `stage`: one of `historical`, `ppt_master`, `ppt_refresh`, `ppt_post_distribution`
+- `enabled`: `true`/`false`
+
+Example:
+
+```yaml
+output_generators:
+  - name: historical_workbook
+    registration: builtin:historical_workbook
+    stage: historical
+    enabled: true
+  - name: ppt_screenshot
+    registration: builtin:ppt_screenshot
+    stage: ppt_master
+    enabled: true
+  - name: ppt_link_refresh
+    registration: builtin:ppt_link_refresh
+    stage: ppt_refresh
+    enabled: false
+  - name: custom_exhibit
+    registration: my_package.outputs:CustomExhibitGenerator
+    stage: ppt_post_distribution
+    enabled: true
+```
+
+Setting `enabled: false` cleanly skips that generator for the run.
+
 ## Start here
 
 - **Project agent guide:** [AGENTS.md](AGENTS.md)

src/counter_risk/config.py

@@ -28,6 +28,50 @@ class InputDiscoveryConfig(BaseModel):
     naming_patterns: dict[str, list[str]] = Field(default_factory=dict)
 
 
+class OutputGeneratorConfig(BaseModel):
+    """Registration record for one output generator."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    name: str
+    registration: str
+    stage: Literal["historical", "ppt_master", "ppt_refresh", "ppt_post_distribution"]
+    enabled: bool = True
+
+    @field_validator("name", "registration")
+    @classmethod
+    def _validate_non_blank(cls, value: str) -> str:
+        normalized = value.strip()
+        if not normalized:
+            raise ValueError("Value must be non-empty")
+        return normalized
+
+
+def _default_output_generators() -> tuple[OutputGeneratorConfig, ...]:
+    return (
+        OutputGeneratorConfig(
+            name="historical_workbook",
+            registration="builtin:historical_workbook",
+            stage="historical",
+        ),
+        OutputGeneratorConfig(
+            name="ppt_screenshot",
+            registration="builtin:ppt_screenshot",
+            stage="ppt_master",
+        ),
+        OutputGeneratorConfig(
+            name="ppt_link_refresh",
+            registration="builtin:ppt_link_refresh",
+            stage="ppt_refresh",
+        ),
+        OutputGeneratorConfig(
+            name="pdf_export",
+            registration="builtin:pdf_export",
+            stage="ppt_post_distribution",
+        ),
+    )
+
+
 class WorkflowConfig(BaseModel):
     """Configuration for a single Counter Risk workflow execution.
 
@@ -56,6 +100,9 @@ class WorkflowConfig(BaseModel):
     distribution_static: bool = False
     export_pdf: bool = False
     enable_ppt_output: bool = True
+    output_generators: tuple[OutputGeneratorConfig, ...] = Field(
+        default_factory=_default_output_generators
+    )
     output_root: Path = Path("runs")
 
     @property
@@ -83,6 +130,19 @@ def _validate_optional_iso_date(cls, value: Any) -> Any:
                 raise ValueError("Value must be a valid ISO date (YYYY-MM-DD)") from exc
         raise ValueError("Value must be a valid ISO date (YYYY-MM-DD)")
 
+    @field_validator("output_generators")
+    @classmethod
+    def _validate_unique_output_generator_names(
+        cls, value: tuple[OutputGeneratorConfig, ...]
+    ) -> tuple[OutputGeneratorConfig, ...]:
+        seen: set[str] = set()
+        for entry in value:
+            normalized_name = entry.name.casefold()
+            if normalized_name in seen:
+                raise ValueError(f"output_generators contains duplicate name: {entry.name!r}")
+            seen.add(normalized_name)
+        return value
+
 
 def _format_validation_error(error: ValidationError) -> str:
     lines = ["Configuration validation failed:"]

src/counter_risk/outputs/__init__.py

@@ -0,0 +1,29 @@
+"""Pluggable output generator interfaces and implementations."""
+
+from .base import OutputContext, OutputGenerator
+from .historical_workbook import (
+    HistoricalWalWorkbookOutputGenerator,
+    HistoricalWorkbookOutputGenerator,
+)
+from .pdf_export import PDFExportGenerator
+from .ppt_link_refresh import (
+    PptLinkRefreshOutputGenerator,
+    PptLinkRefreshResult,
+    PptLinkRefreshStatus,
+)
+from .ppt_screenshot import PptScreenshotOutputGenerator
+from .registry import OutputGeneratorRegistry, OutputGeneratorRegistryContext
+
+__all__ = [
+    "HistoricalWalWorkbookOutputGenerator",
+    "HistoricalWorkbookOutputGenerator",
+    "OutputContext",
+    "OutputGenerator",
+    "OutputGeneratorRegistry",
+    "OutputGeneratorRegistryContext",
+    "PDFExportGenerator",
+    "PptLinkRefreshOutputGenerator",
+    "PptLinkRefreshResult",
+    "PptLinkRefreshStatus",
+    "PptScreenshotOutputGenerator",
+]

src/counter_risk/outputs/base.py

@@ -0,0 +1,30 @@
+"""Output generator interface for pluggable pipeline outputs."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import date
+from pathlib import Path
+from typing import Protocol
+
+from counter_risk.config import WorkflowConfig
+
+
+@dataclass(frozen=True)
+class OutputContext:
+    """Immutable run context passed to output generators."""
+
+    config: WorkflowConfig
+    run_dir: Path
+    as_of_date: date
+    run_date: date
+    warnings: tuple[str, ...] = field(default_factory=tuple)
+
+
+class OutputGenerator(Protocol):
+    """Interface implemented by output plugins."""
+
+    name: str
+
+    def generate(self, *, context: OutputContext) -> tuple[Path, ...]:
+        """Generate one output type and return created output paths."""

src/counter_risk/outputs/historical_workbook.py

@@ -0,0 +1,82 @@
+"""Output generators for historical workbook updates."""
+
+from __future__ import annotations
+
+import shutil
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any, cast
+
+from counter_risk.calculations.wal import calculate_wal
+from counter_risk.outputs.base import OutputContext, OutputGenerator
+from counter_risk.writers.historical_update import append_wal_row, locate_ex_llc_3_year_workbook
+
+_WorkbookLocator = Callable[[], Path]
+_WalCalculator = Callable[[Path, date], float]
+_WalAppender = Callable[..., Path]
+_WorkbookCopier = Callable[[str | Path, str | Path], str]
+_HistoricalWorkbookMerger = Callable[..., None]
+_RecordsExtractor = Callable[[Any], list[dict[str, Any]]]
+
+
+@dataclass(frozen=True)
+class HistoricalWorkbookOutputGenerator(OutputGenerator):
+    """Generate historical workbook outputs for all reporting variants."""
+
+    parsed_by_variant: Mapping[str, Mapping[str, Any]]
+    warnings: list[str]
+    workbook_merger: _HistoricalWorkbookMerger
+    records_extractor: _RecordsExtractor
+    name: str = "historical_workbook"
+    workbook_copier: _WorkbookCopier = cast(_WorkbookCopier, shutil.copy2)
+
+    def generate(self, *, context: OutputContext) -> tuple[Path, ...]:
+        mosers_all_programs = context.config.mosers_all_programs_xlsx
+        if mosers_all_programs is None:
+            raise ValueError("mosers_all_programs_xlsx is required for pipeline execution")
+
+        variant_inputs = (
+            ("all_programs", mosers_all_programs, context.config.hist_all_programs_3yr_xlsx),
+            ("ex_trend", context.config.mosers_ex_trend_xlsx, context.config.hist_ex_llc_3yr_xlsx),
+            ("trend", context.config.mosers_trend_xlsx, context.config.hist_llc_3yr_xlsx),
+        )
+
+        output_paths: list[Path] = []
+        for variant, _workbook_path, historical_path in variant_inputs:
+            target_hist = context.run_dir / historical_path.name
+            self.workbook_copier(historical_path, target_hist)
+            totals_records = self.records_extractor(self.parsed_by_variant[variant]["totals"])
+            self.workbook_merger(
+                workbook_path=target_hist,
+                variant=variant,
+                as_of_date=context.as_of_date,
+                totals_records=totals_records,
+                warnings=self.warnings,
+            )
+            output_paths.append(target_hist)
+
+        return tuple(output_paths)
+
+
+@dataclass(frozen=True)
+class HistoricalWalWorkbookOutputGenerator(OutputGenerator):
+    """Generate the historical WAL workbook update output."""
+
+    exposure_summary_path: Path
+    name: str = "historical_wal_workbook"
+    workbook_path: Path | None = None
+    workbook_locator: _WorkbookLocator = locate_ex_llc_3_year_workbook
+    wal_calculator: _WalCalculator = calculate_wal
+    wal_appender: _WalAppender = append_wal_row
+
+    def generate(self, *, context: OutputContext) -> tuple[Path, ...]:
+        workbook_path = self.workbook_path or self.workbook_locator()
+        wal_value = self.wal_calculator(self.exposure_summary_path, context.as_of_date)
+        updated_workbook = self.wal_appender(
+            workbook_path,
+            px_date=context.as_of_date,
+            wal_value=wal_value,
+        )
+        return (updated_workbook,)

src/counter_risk/outputs/pdf_export.py

@@ -0,0 +1,77 @@
+"""Output generator for distribution PDF export via PowerPoint COM."""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from contextlib import suppress
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from counter_risk.integrations.powerpoint_com import is_powerpoint_com_available
+from counter_risk.outputs.base import OutputContext, OutputGenerator
+
+LOGGER = logging.getLogger(__name__)
+_PDF_FIXED_FORMAT_TYPE = 2  # ppFixedFormatTypePDF
+
+_ComAvailabilityChecker = Callable[[], bool]
+_PptxToPdfExporter = Callable[[Path, Path], None]
+
+
+def export_pptx_to_pdf_via_com(source_pptx: Path, pdf_path: Path) -> None:
+    """Export a PowerPoint deck to PDF using PowerPoint COM automation."""
+    from counter_risk.integrations.powerpoint_com import initialize_powerpoint_application
+
+    app: Any | None = None
+    presentation: Any | None = None
+    try:
+        app = initialize_powerpoint_application()
+        with suppress(Exception):
+            app.Visible = 0
+        presentation = app.Presentations.Open(str(source_pptx), False, True, False)
+        presentation.ExportAsFixedFormat(str(pdf_path), _PDF_FIXED_FORMAT_TYPE)
+    finally:
+        if presentation is not None:
+            with suppress(Exception):
+                presentation.Close()
+        if app is not None:
+            with suppress(Exception):
+                app.Quit()
+
+
+@dataclass(frozen=True)
+class PDFExportGenerator(OutputGenerator):
+    """Generate a distribution PDF output when COM export is available."""
+
+    source_pptx: Path
+    warnings: list[str]
+    name: str = "pdf_export"
+    com_availability_checker: _ComAvailabilityChecker = is_powerpoint_com_available
+    pptx_to_pdf_exporter: _PptxToPdfExporter = export_pptx_to_pdf_via_com
+
+    def generate(self, *, context: OutputContext) -> tuple[Path, ...]:
+        if not context.config.export_pdf:
+            LOGGER.warning("distribution_pdf_skipped reason=export_pdf_disabled")
+            return ()
+
+        if not self.com_availability_checker():
+            warning = (
+                "distribution_pdf requested but PowerPoint COM is unavailable; "
+                "skipping PDF generation"
+            )
+            self.warnings.append(warning)
+            LOGGER.warning(
+                "distribution_pdf_skipped reason=com_unavailable source=%s", self.source_pptx
+            )
+            return ()
+
+        pdf_path = context.run_dir / f"{self.source_pptx.stem}.pdf"
+        try:
+            self.pptx_to_pdf_exporter(self.source_pptx, pdf_path)
+        except Exception as exc:
+            LOGGER.error("PDF export failed: %s", exc)
+            raise RuntimeError(f"PDF export failed: {exc}") from exc
+
+        LOGGER.info("distribution_pdf_complete path=%s", pdf_path)
+        return (pdf_path,)