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, or_, 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,
                        user_id=report.user_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, user_id: str | None = None
    ) -> list[AnalysisReport]:
        """Every version for one analysis, oldest-first.

        `user_id` scopes the read to the owner (2026-07-23) but **tolerates NULL**:
        `ReportStore.save` did not write `reports.user_id` until pr/18 (2026-07-22),
        so every report generated before that date has a NULL owner. A strict
        equality filter would hide them, so legacy rows are matched too — the loosest-
        deployment-shape convention (§7D). Drop the NULL branch once the column is
        backfilled.
        """
        async with AsyncSessionLocal() as session:
            where = [AnalysisReportRow.analysis_id == analysis_id]
            if user_id is not None:
                where.append(
                    or_(
                        AnalysisReportRow.user_id == user_id,
                        AnalysisReportRow.user_id.is_(None),
                    )
                )
            result = await session.execute(
                select(AnalysisReportRow)
                .where(*where)
                .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)