src/agents/report/store.py

"""ReportStore — persists/reads versioned AnalysisReports (KM-644).

Mirrors `PostgresReportInputStore`: each call opens its own `AsyncSessionLocal`.

Version assignment is serialized per `analysis_id` with a Postgres
transaction-level advisory lock so concurrent button presses can't compute the
same version number; the `(analysis_id, version)` unique constraint is the
backstop. Per decision 4A every generation is a new version, so two
near-simultaneous presses legitimately produce V<n> and V<n+1> — the lock only
prevents a duplicate-number race, not double generation.
"""

from __future__ import annotations

import hashlib

from sqlalchemy import func, select, text

from src.db.postgres.connection import AsyncSessionLocal
from src.db.postgres.models import AnalysisReportRow
from src.middlewares.logging import get_logger

from .schemas import AnalysisReport

logger = get_logger("report_store")


def _lock_key(analysis_id: str) -> int:
    """Stable signed 64-bit key for `pg_advisory_xact_lock`.

    Python's builtin `hash(str)` is randomized per process, so derive a
    deterministic key from a digest instead.
    """
    digest = hashlib.sha256(analysis_id.encode()).digest()
    return int.from_bytes(digest[:8], "big", signed=True)


def _report_title(report: AnalysisReport) -> str:
    """Title for the dedorch `reports.title` column — the goal, else a generic label."""
    objective = (report.problem_statement.objective or "").strip()
    return objective[:200] if objective else "Analysis Report"


def _row_to_report(row) -> AnalysisReport:
    """Rebuild a minimal AnalysisReport from the flat dedorch row.

    dedorch stores markdown only, so structured fields (findings/caveats/…) come back
    empty; `rendered_markdown` carries the content the FE renders/downloads.
    """
    return AnalysisReport(
        report_id=row.id,
        analysis_id=row.analysis_id,
        version=row.version,
        generated_at=row.generated_at,
        rendered_markdown=row.content,
    )


class ReportStore:
    """Read/write versioned reports keyed by `analysis_id`."""

    async def save(self, report: AnalysisReport) -> AnalysisReport:
        """Assign the next version under an advisory lock and persist.

        Mutates and returns `report` with its final `version`.
        """
        async with AsyncSessionLocal() as session:
            async with session.begin():
                await session.execute(
                    text("SELECT pg_advisory_xact_lock(:k)"),
                    {"k": _lock_key(report.analysis_id)},
                )
                result = await session.execute(
                    select(func.max(AnalysisReportRow.version)).where(
                        AnalysisReportRow.analysis_id == report.analysis_id
                    )
                )
                report.version = (result.scalar_one_or_none() or 0) + 1
                session.add(
                    AnalysisReportRow(
                        id=report.report_id,
                        analysis_id=report.analysis_id,
                        title=_report_title(report),
                        content=report.rendered_markdown or "",
                        generated_at=report.generated_at,
                        version=report.version,
                    )
                )
            # leaving session.begin() commits, which releases the advisory lock
        logger.info(
            "report persisted",
            analysis_id=report.analysis_id,
            version=report.version,
            report_id=report.report_id,
        )
        return report

    async def list_for_analysis(self, analysis_id: str) -> list[AnalysisReport]:
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                select(AnalysisReportRow)
                .where(AnalysisReportRow.analysis_id == analysis_id)
                .order_by(AnalysisReportRow.version.asc())
            )
            rows = result.scalars().all()
        return [_row_to_report(row) for row in rows]

    async def get(self, analysis_id: str, version: int) -> AnalysisReport | None:
        async with AsyncSessionLocal() as session:
            result = await session.execute(
                select(AnalysisReportRow).where(
                    AnalysisReportRow.analysis_id == analysis_id,
                    AnalysisReportRow.version == version,
                )
            )
            row = result.scalar_one_or_none()
        if row is None:
            return None
        return _row_to_report(row)