diff --git a/backend/ipo/__init__.py b/backend/ipo/__init__.py index d36a79b..983a950 100644 --- a/backend/ipo/__init__.py +++ b/backend/ipo/__init__.py @@ -134,6 +134,7 @@ from backend.ipo.scoring.service import ( SCREENER_MODEL_VERSION, IpoRescoreOutcome, + get_current_evaluation, rescore_issue, ) from backend.ipo.sources.enrichment import ( @@ -229,6 +230,7 @@ "evaluate_caution_flags", "evaluate_issue", "fetch_sebi_filings", + "get_current_evaluation", "get_document", "get_evaluation", "get_financial", diff --git a/backend/ipo/dashboard.py b/backend/ipo/dashboard.py index 849e42a..e46edc7 100644 --- a/backend/ipo/dashboard.py +++ b/backend/ipo/dashboard.py @@ -29,8 +29,6 @@ ) from backend.ipo.repository import ( SessionFactory, - get_latest_evaluation, - get_latest_manual_profile, list_documents, list_enrichment_signals, list_extraction_proposals, @@ -38,6 +36,8 @@ list_subscriptions, ) from backend.ipo.scoring.score_model import PDF_WEIGHTS +from backend.ipo.scoring.service import get_current_evaluation +from backend.ipo.scoring.state import normalize_scoring_time from backend.storage import session_scope # Selection thresholds for the strengths/risks columns: a factor earning at @@ -80,6 +80,8 @@ class IpoDashboardRow: documents_total: int breakdown: tuple[ScoreBreakdownItem, ...] = () evaluation_stale: bool = False + calculated_at: dt.datetime | None = None + last_verified_at: dt.datetime | None = None @dataclass(frozen=True) @@ -133,23 +135,39 @@ def top_positive_and_risk_reasons( def _row_for_issue( - issue: Any, *, session_factory: SessionFactory + issue: Any, *, now: dt.datetime, session_factory: SessionFactory ) -> IpoDashboardRow: """Denormalize one issue's stored state into a display-ready row. + Args: + issue: Issue identity from the inventory; scoring fields are reread together. + now: Shared aware UTC render instant for semantic time-eligibility checks. + session_factory: Caller-owned repository transaction factory. + + Returns: + Current or explicitly stale historical content with separate calculation, + verification and overall display-activity timestamps. + + Raises: + IpoScoringConflictError: If all three current-read assembly attempts conflict. + IpoNotFoundError: If the issue disappears during current or display reads. + ValueError: If now has no timezone offset. + Beginner note: ``last_updated`` considers every evidence source, while - ``evaluation_stale`` asks the narrower question: did any evidence - change after the displayed score was computed? Keeping both concepts + ``evaluation_stale`` asks the narrower question: does the revision, model and + semantic fingerprint still match at this render time? Keeping both concepts explicit prevents a fresh-looking timestamp from hiding an old verdict. """ + current = get_current_evaluation(issue.id, as_of=now, session_factory=session_factory) + issue = current.snapshot.inputs.issue + profile = current.snapshot.inputs.profile documents = [ document for document in list_documents(issue.id, session_factory=session_factory) if document.document_type in {"drhp", "rhp"} ] downloaded = sum(1 for document in documents if document.content_sha256) - profile = get_latest_manual_profile(issue.id, session_factory=session_factory) proposals = list_extraction_proposals( issue_id=issue.id, session_factory=session_factory, @@ -164,24 +182,8 @@ def _row_for_issue( enrichment = list_enrichment_signals( issue.id, session_factory=session_factory ) - evaluation = get_latest_evaluation(issue.id, session_factory=session_factory) - source_documents = tuple( - dict.fromkeys( - ( - *( - document.document_url - for document in documents - if getattr(document, "document_url", None) - ), - *( - (profile.source_document_url,) - if profile is not None - and getattr(profile, "source_document_url", None) - else () - ), - ) - ) - ) + evaluation = current.evaluation + source_documents = current.snapshot.inputs.source_documents evidence_times = [ value for value in ( @@ -219,16 +221,13 @@ def _row_for_issue( if value is not None ] latest_evidence_at = max(evidence_times, default=None) - evaluation_stale = bool( - evaluation is not None - and latest_evidence_at is not None - and latest_evidence_at > evaluation.scored_at - ) + evaluation_stale = not current.fresh last_updated = max( ( value for value in ( latest_evidence_at, + current.last_verified_at, evaluation.scored_at if evaluation is not None else None, ) if value is not None @@ -288,6 +287,8 @@ def _row_for_issue( documents_total=len(documents), breakdown=result.breakdown, evaluation_stale=evaluation_stale, + calculated_at=evaluation.scored_at, + last_verified_at=current.last_verified_at, ) @@ -298,15 +299,29 @@ def build_dashboard_snapshot( ) -> IpoDashboardSnapshot: """Read every issue's stored state into one display-ready snapshot. + Args: + now: Optional aware render clock; defaults to the current UTC instant. + session_factory: Transaction factory used only for repository reads. + + Returns: + Immutable rows classified at the same business-time instant. + + Raises: + ValueError: If now has no timezone offset; supply an aware instant. + IpoScoringConflictError: If any issue exhausts the current reader's three + assembly attempts; retry the render later without certifying old data. + IpoNotFoundError: If an issue is deleted after inventory is listed; refresh + the inventory rather than treating its old verdict as current. + Beginner note: The per-issue reads are simple repository calls rather than one big join because the IPO universe is dozens of issues, not thousands; the - page additionally caches the snapshot, so clarity wins over query - golf here. + page rebuilds actionable snapshots each render with one UTC clock. + A TTL alone cannot detect another session writing or GMP expiring. """ - when = now if now is not None else dt.datetime.now(dt.UTC) + when = normalize_scoring_time(now) if now is not None else dt.datetime.now(dt.UTC) rows = tuple( - _row_for_issue(issue, session_factory=session_factory) + _row_for_issue(issue, now=when, session_factory=session_factory) for issue in list_issues(session_factory=session_factory) ) return IpoDashboardSnapshot(generated_at=when, rows=rows) @@ -361,13 +376,13 @@ def section_recommended(snapshot: IpoDashboardSnapshot) -> tuple[IpoDashboardRow Beginner note: Filtering uses the stored evaluation only. It never recalculates a - score, so stale evaluations stay visible and are also routed to the - review queue for an explicit re-score. + score. Stale evaluations remain visible in history/all-issues only, + and enter the review queue for an explicit re-score. """ return tuple( row for row in snapshot.rows - if row.recommendation == Recommendation.RECOMMENDED.value + if not row.evaluation_stale and row.recommendation == Recommendation.RECOMMENDED.value ) @@ -382,7 +397,7 @@ def section_not_recommended(snapshot: IpoDashboardSnapshot) -> tuple[IpoDashboar return tuple( row for row in snapshot.rows - if row.recommendation == Recommendation.NOT_RECOMMENDED.value + if not row.evaluation_stale and row.recommendation == Recommendation.NOT_RECOMMENDED.value ) diff --git a/backend/ipo/repository.py b/backend/ipo/repository.py index de21743..6e06c81 100644 --- a/backend/ipo/repository.py +++ b/backend/ipo/repository.py @@ -82,6 +82,12 @@ ) from backend.ipo.scoring.recommendation import build_recommendation from backend.ipo.scoring.score_model import score_ipo +from backend.ipo.scoring.state import ( + IpoScoringConflictError, + IpoScoringSnapshot, + IpoScoringStateRecord, + normalize_scoring_time, +) from backend.observability import ( EVENT_IPO_DOCUMENT_DOWNLOAD_COMPLETED, EVENT_IPO_DOCUMENT_DOWNLOAD_FAILED, @@ -91,7 +97,7 @@ ) from backend.scanning.result_contract import normalize_secret_safe_json from backend.security import redact_text -from backend.storage import session_scope +from backend.storage import SessionFactory, session_scope from backend.storage.ipo_repository import ( delete_ipo_document_row, delete_ipo_evaluation_row, @@ -108,6 +114,7 @@ get_ipo_issue, get_ipo_issue_by_sebi_key, get_ipo_manual_extraction, + get_ipo_scoring_state_values, get_ipo_subscription, get_latest_ipo_evaluation_rows, get_latest_ipo_filing_date, @@ -120,6 +127,7 @@ insert_ipo_issue, insert_ipo_manual_extraction, insert_ipo_subscription, + ipo_evaluation_payload_matches, list_ipo_document_rows, list_ipo_enrichment_signal_rows, list_ipo_evaluation_rows, @@ -129,7 +137,9 @@ list_ipo_manual_extraction_rows, list_ipo_subscription_rows, list_unclaimed_ipo_issues_by_company_name, + lock_ipo_scoring_state, mark_ipo_extraction_proposal_reviewed, + select_ipo_current_evaluation, try_insert_ipo_extraction_proposal, update_ipo_document_cache_if_source_matches, update_ipo_document_values, @@ -139,7 +149,6 @@ upsert_ipo_enrichment_signal, ) -SessionFactory = Any DocumentDownloader = Callable[..., IpoDocumentDownloadResult] AuditRecorder = Callable[..., bool] @@ -914,29 +923,46 @@ def get_latest_ipo_ratios( ) -def load_ipo_factor_inputs_snapshot( +def load_ipo_scoring_snapshot( issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = session_scope, -) -> IpoFactorInputs: - """Load every scoring input in one caller-owned read transaction. +) -> IpoScoringSnapshot: + """Detach one revision-consistent evidence and current-selection bundle. + + Args: + issue_id: Issue whose evidence and selected history are required. + as_of: Aware business clock, normalized to UTC before time-derived rules. + session_factory: Fresh caller-owned read transaction for this attempt. + + Returns: + Detached inputs, scalar state and a complete issue-owned historical pair. + + Raises: + IpoNotFoundError: If the issue does not exist. + IpoScoringConflictError: If committed writes changed the bundle. + ValueError: If the injected clock is naive. Beginner note: - A scheduled job must not combine a profile from one instant with a - subscription or enrichment row committed a moment later. Detaching the - complete bundle inside one transaction makes the fingerprint and score - consume the same immutable snapshot. + One READ COMMITTED transaction alone is not a consistent snapshot. + Compare scalar state SQL reads around all input and eager-child reads; + ORM identity-map values cannot provide the second read. This function + does exactly one attempt so the service owns one three-attempt budget. """ from backend.ipo.scoring.factor_derivation import ( IpoFactorInputs, derive_debt_reduction_purpose_evidence, ) + as_of = normalize_scoring_time(as_of) with session_factory() as session: + before = get_ipo_scoring_state_values(session, issue_id) issue_row = get_ipo_issue(session, issue_id) if issue_row is None: raise IpoNotFoundError(f"IPO issue {issue_id} was not found.") + if before is None: + raise IpoScoringConflictError("IPO scoring state is missing.") profile_row = get_latest_ipo_manual_extraction(session, issue_id) subscription_row = get_latest_ipo_subscription(session, issue_id) enrichment_rows = list_ipo_enrichment_signal_rows(session, issue_id) @@ -958,6 +984,14 @@ def load_ipo_factor_inputs_snapshot( if row.document_type in {"drhp", "rhp"} ) ) + selected = ( + get_ipo_evaluation_rows(session, issue_id, before[1]) + if before[1] is not None else get_latest_ipo_evaluation_rows(session, issue_id) + ) + evaluation = _evaluation_record(*selected) if selected is not None else None + after = get_ipo_scoring_state_values(session, issue_id) + if before != after: + raise IpoScoringConflictError("IPO inputs changed during snapshot assembly.") ratios = ( calculate_ipo_ratios( profile, @@ -967,7 +1001,7 @@ def load_ipo_factor_inputs_snapshot( if profile is not None else None ) - return IpoFactorInputs( + inputs = IpoFactorInputs( issue=issue, profile=profile, ratios=ratios, @@ -978,6 +1012,37 @@ def load_ipo_factor_inputs_snapshot( source_documents=source_documents, ) + return IpoScoringSnapshot(inputs, IpoScoringStateRecord(*before), evaluation) + + +def load_ipo_factor_inputs_snapshot( + issue_id: int, *, as_of: dt.datetime, session_factory: SessionFactory = session_scope +) -> IpoFactorInputs: + """Keep the historical input-only API with revision-validated assembly. + + Args: + issue_id: Issue whose complete scoring inputs should be detached. + as_of: Aware business clock, normalized to UTC for eligibility rules. + session_factory: Caller-owned read transaction factory for this attempt. + + Returns: + Detached input bundle whose scalar revision was unchanged around all + component and eager-child reads; selection metadata is not returned. + + Raises: + IpoNotFoundError: If the issue does not exist or disappears during assembly. + IpoScoringConflictError: If the state is missing or changes during this + single attempt; callers may reload later without assuming success. + ValueError: If as_of has no timezone offset. + + Beginner note: + Conflicts propagate to the caller; this compatibility wrapper introduces + no nested retry budget and never writes verification state. Unlike the + current-evaluation reader's three-attempt policy, this loader makes one + attempt so orchestration can share its budget with publication conflicts. + """ + return load_ipo_scoring_snapshot(issue_id, as_of=as_of, session_factory=session_factory).inputs + _STATUS_ORDER = { IpoStatus.DRHP_FILED: 0, @@ -2437,21 +2502,47 @@ def _evaluate_issue_once( inputs_fingerprint: str | None = None, model_version: str = "ipo-001-v1", session_factory: SessionFactory = session_scope, + expected_revision: int | None = None, + publication_check: Callable[[], None] | None = None, ) -> tuple[IpoEvaluationRecord, bool]: - """Persist one semantic evaluation once and report whether this call won. + """Persist one immutable evaluation and optionally publish verified selection. + + Args: + issue_id: Owner of all supplied source documents and resulting receipts. + score_input: Derived scorecard; public legacy callers remain history-only. + caution_flags: Optional caution receipt used by recommendation derivation. + inputs_fingerprint: Semantic identity of the complete detached inputs. + model_version: Version persisted without modifying older history. + session_factory: Caller-owned write scope for this single attempt. + expected_revision: Service-only captured revision enabling guarded publication. + publication_check: Optional service callback rejecting clock eligibility drift. + + Returns: + The complete immutable pair and whether a new pair was inserted. + + Raises: + IpoNotFoundError: If the issue no longer exists. + IpoScoringConflictError: If revision or live time eligibility changed. + IpoValidationError: If company/source ownership fails or an existing pair's + payload differs from the complete freshly derived candidate. Beginner note: - The three IPO-006 keyword arguments are optional so IPO-001 callers - keep their exact behavior. The scoring service passes a caution-flag - report (enforced inside ``build_recommendation``), the SHA-256 - fingerprint of the evidence it consumed (the screener's idempotency - anchor), and its own model version; all three are persisted with the - immutable pair. + Conditional state UPDATE precedes subordinate reads and nested insertion, + holding the lock through commit. It establishes SQLite's outer transaction + so a late callback/selection failure also rolls back the saved pair. The + public evaluate_issue API never passes revision authority and cannot certify + caller-invented scorecards. Retry ownership stays with the scoring service. """ score_result = score_ipo(score_input) recommendation = build_recommendation(score_result, caution_flags=caution_flags) with session_factory() as session: + if expected_revision is not None and not lock_ipo_scoring_state( + session, issue_id, expected_revision=expected_revision + ): + if get_ipo_issue(session, issue_id) is None: + raise IpoNotFoundError(f"IPO issue {issue_id} was not found.") + raise IpoScoringConflictError("IPO inputs changed before publication.") issue = get_ipo_issue(session, issue_id) if issue is None: raise IpoNotFoundError(f"IPO issue {issue_id} was not found.") @@ -2522,6 +2613,25 @@ def _evaluate_issue_once( score_row, recommendation_row, inserted = insert_ipo_evaluation( session, issue_id, score_values, recommendation_values ) + if expected_revision is not None: + # A legacy caller can claim the same fingerprint for arbitrary factors. + # Check full content under the lock before granting that row authority; + # raising rolls back this attempt without rewriting history or state. + if not inserted and not ipo_evaluation_payload_matches( + score_row, + recommendation_row, + score_values=score_values, + recommendation_values=recommendation_values, + ): + raise IpoValidationError("Stored IPO evaluation payload does not match current derived evidence.") + # Check the clock after insertion too: crossing eligibility must + # roll back the complete outer transaction, including its savepoint. + if publication_check is not None: + publication_check() + select_ipo_current_evaluation( + session, issue_id, score_row.id, input_revision=expected_revision, + verified_at=dt.datetime.now(dt.UTC), + ) return _evaluation_record(score_row, recommendation_row), inserted @@ -2586,9 +2696,9 @@ def get_latest_evaluation( ) -> IpoEvaluationRecord | None: """Return the newest complete evaluation record for one issue, if any. - The IPO-006 scoring service compares its freshly computed inputs - fingerprint against this record to decide whether a re-score would be a - byte-identical no-op, which is what makes ``run_ipo_screener`` idempotent. + This is explicitly historical ordering, not the currently verified receipt. + Actionable consumers must use ``get_current_evaluation`` from the scoring + service, because returning to older semantic inputs may reselect older history. Beginner note: “Latest” is a display convenience, not an in-place update: every @@ -2631,11 +2741,13 @@ def get_latest_subscription( def get_latest_recommendation( issue_id: int, *, session_factory: SessionFactory = session_scope ) -> IpoRecommendationResult | None: - """Return the newest recommendation for an issue, or ``None`` if unscored. + """Return the newest historical recommendation, or ``None`` if unscored. Reads only the most recent evaluation pair (``LIMIT 1``) rather than loading the full append-only history. A missing issue still raises ``IpoNotFoundError`` so callers can distinguish "no such issue" from "issue exists but unscored". + It does not certify current freshness; use ``get_current_evaluation`` for + actionable decisions. """ with session_factory() as session: if get_ipo_issue(session, issue_id) is None: diff --git a/backend/ipo/scoring/service.py b/backend/ipo/scoring/service.py index 8ca8b50..29d69e5 100644 --- a/backend/ipo/scoring/service.py +++ b/backend/ipo/scoring/service.py @@ -7,14 +7,12 @@ the dashboard's re-score button call :func:`rescore_issue`, so a manual click and a scheduled run can never disagree about how scoring works. -Beginner note — how idempotency works here: -Before persisting, the service computes a SHA-256 *inputs fingerprint* over -exactly the evidence and rule versions it consumed. If the newest stored -evaluation was produced by the same model version from the same fingerprint, -re-scoring would write a byte-identical row, so the service reports -``skipped_unchanged`` instead. Re-running the screener is therefore free -until some real input (a new revision, price band, subscription snapshot, or -enrichment observation) actually changes. +Beginner note: +A semantic fingerprint identifies immutable content across all history, while a +separate current-state row records the input revision and wall time verified by +publication. A -> B -> A reuses historical A without changing its receipt or +calculation time. Revision checks reject concurrent evidence changes; semantic +freshness also checks model and time eligibility on each actionable read. """ from __future__ import annotations @@ -23,7 +21,7 @@ import hashlib import json import logging -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Final, Literal from backend.ipo.models import ( @@ -35,8 +33,7 @@ from backend.ipo.repository import ( SessionFactory, _evaluate_issue_once, - get_latest_evaluation, - load_ipo_factor_inputs_snapshot, + load_ipo_scoring_snapshot, ) from backend.ipo.scoring.caution_flags import ( CAUTION_FLAGS_VERSION, @@ -49,12 +46,13 @@ IpoFactorInputs, derive_score_input, ) +from backend.ipo.scoring.state import IpoCurrentEvaluation, IpoScoringConflictError, normalize_scoring_time from backend.observability import EVENT_IPO_ISSUE_SCORED, log_event from backend.storage import session_scope logger = logging.getLogger(__name__) -SCREENER_MODEL_VERSION: Final = "ipo-006-v2" +SCREENER_MODEL_VERSION: Final = "ipo-013-v3" @dataclass(frozen=True) @@ -81,6 +79,12 @@ class IpoRescoreOutcome: def compute_inputs_fingerprint(inputs: IpoFactorInputs) -> str: """Hash exactly the evidence and rule versions scoring will consume. + Args: + inputs: Detached, revision-consistent evidence with an aware UTC clock. + + Returns: + SHA-256 identity covering semantic evidence and canonical source URLs. + Beginner note: Two time-derived facts are hashed instead of the clock itself: the set of GMP observations still inside the staleness window, and whether the @@ -264,6 +268,7 @@ def compute_inputs_fingerprint(inputs: IpoFactorInputs) -> str: else None ), "near_close": near_close, + "source_documents": sorted(set(inputs.source_documents)), } encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(encoded.encode("utf-8")).hexdigest() @@ -288,71 +293,123 @@ def rescore_issue( An outcome whose status says whether a new evaluation was persisted, an identical one already existed, or the evidence was insufficient. + Raises: + IpoNotFoundError: If the issue disappears before successful publication. + IpoScoringConflictError: If the three total attempts cannot verify evidence. + IpoValidationError: If a historical pair claims the current fingerprint + but its scoring or recommendation payload differs from the derived + candidate. No current selection or immutable history is changed. + ValueError: If the injected business clock is naive. + Beginner note: + Snapshot and publication share at most three total attempts. Each + conflict rolls back its complete write scope before a fresh read; + unrelated persistence errors are not disguised as retryable conflicts. + Reusing old A after A -> B -> A updates verification, never history. + A legacy caller can supply a fingerprint, so matching identity alone + cannot authorize reuse: the complete stored receipt must agree as well. No network happens here — every input is a repository read, so the dashboard's re-score button can safely call this inside a page action. """ - when = as_of if as_of is not None else dt.datetime.now(dt.UTC) - inputs = load_ipo_factor_inputs_snapshot( - issue_id, - as_of=when, - session_factory=session_factory, - ) - issue = inputs.issue - if inputs.profile is None: - return IpoRescoreOutcome( - issue_id=issue_id, - company_name=issue.company_name, - status="insufficient_inputs", - missing=("manual_extraction",), - ) + frozen = normalize_scoring_time(as_of) if as_of is not None else None + for _attempt in range(3): + try: + when = frozen if frozen is not None else _utc_now() + snapshot = load_ipo_scoring_snapshot(issue_id, as_of=when, session_factory=session_factory) + inputs = snapshot.inputs + issue = inputs.issue + if inputs.profile is None: + return IpoRescoreOutcome( + issue_id=issue_id, company_name=issue.company_name, + status="insufficient_inputs", missing=("manual_extraction",), + ) + fingerprint = compute_inputs_fingerprint(inputs) - fingerprint = compute_inputs_fingerprint(inputs) + def check_publication_clock( + inputs: IpoFactorInputs = inputs, fingerprint: str = fingerprint + ) -> None: + """Reject eligibility drift without hashing raw wall-clock time. - latest = get_latest_evaluation(issue_id, session_factory=session_factory) - if ( - latest is not None - and latest.model_version == SCREENER_MODEL_VERSION - and latest.inputs_fingerprint == fingerprint - ): - return IpoRescoreOutcome( - issue_id=issue_id, - company_name=issue.company_name, - status="skipped_unchanged", - evaluation=latest, - ) + Beginner note: + Frozen test/business clocks intentionally stay fixed. A live + calculation must retry if GMP expiry or UTC near-close date + changes while deriving/persisting, even without a DB write. + """ + if frozen is None and compute_inputs_fingerprint(replace(inputs, as_of=_utc_now())) != fingerprint: + raise IpoScoringConflictError("IPO time eligibility changed before publication.") - score_input = derive_score_input(inputs) - caution_flags = evaluate_caution_flags(inputs) - evaluation, inserted = _evaluate_issue_once( - issue_id, - score_input, - caution_flags=caution_flags, - inputs_fingerprint=fingerprint, - model_version=SCREENER_MODEL_VERSION, - session_factory=session_factory, - ) - if not inserted: + evaluation, inserted = _evaluate_issue_once( + issue_id, derive_score_input(inputs), caution_flags=evaluate_caution_flags(inputs), + inputs_fingerprint=fingerprint, model_version=SCREENER_MODEL_VERSION, + session_factory=session_factory, expected_revision=snapshot.state.input_revision, + publication_check=check_publication_clock, + ) + except IpoScoringConflictError: + # The failed session scope has rolled back before the next attempt. + # Snapshot and publication conflicts consume the SAME total budget. + continue + if inserted: + log_event( + logger, EVENT_IPO_ISSUE_SCORED, issue_id=issue_id, + score=str(evaluation.result.score), recommendation=evaluation.result.recommendation.value, + recommendation_type=evaluation.result.recommendation_type, + triggered_flags=sum(flag.status.value == "triggered" for flag in evaluation.result.caution_flags), + ) return IpoRescoreOutcome( - issue_id=issue_id, - company_name=issue.company_name, - status="skipped_unchanged", - evaluation=evaluation, + issue_id=issue_id, company_name=issue.company_name, + status="evaluated" if inserted else "skipped_unchanged", evaluation=evaluation, ) - log_event( - logger, - EVENT_IPO_ISSUE_SCORED, - issue_id=issue_id, - score=str(evaluation.result.score), - recommendation=evaluation.result.recommendation.value, - recommendation_type=evaluation.result.recommendation_type, - triggered_flags=len( - [flag for flag in evaluation.result.caution_flags if flag.status.value == "triggered"] - ), - ) - return IpoRescoreOutcome( - issue_id=issue_id, - company_name=issue.company_name, - status="evaluated", - evaluation=evaluation, - ) + raise IpoScoringConflictError("IPO scoring inputs changed during all three attempts; retry later.") + + +def _utc_now() -> dt.datetime: + """Return the live UTC eligibility clock, separable from frozen business time.""" + return dt.datetime.now(dt.UTC) + + +def get_current_evaluation( + issue_id: int, *, as_of: dt.datetime | None = None, session_factory: SessionFactory = session_scope +) -> IpoCurrentEvaluation: + """Read semantic freshness without scoring, publishing, or contacting providers. + + Args: + issue_id: Issue whose current selection should be checked. + as_of: Aware render time; one shared instant should serve a whole page. + session_factory: Fresh read transaction for each bounded attempt. + + Returns: + Owned historical content and an explicit actionable freshness decision. + + Raises: + IpoScoringConflictError: After three snapshot conflicts. + IpoNotFoundError: If the issue is missing. + ValueError: If as_of is naive. + + Beginner note: + Matching semantic content is necessary but cannot certify an unchecked + input revision. Neither a read nor a cache hit writes verification time. + Time and model changes can make a selected score stale without DB writes. + """ + when = normalize_scoring_time(as_of) if as_of is not None else _utc_now() + for _attempt in range(3): + try: + snapshot = load_ipo_scoring_snapshot(issue_id, as_of=when, session_factory=session_factory) + except IpoScoringConflictError: + continue + evaluation, state = snapshot.evaluation, snapshot.state + reason = "current" + if ( + evaluation is None or state.current_score_id is None + or state.last_verified_at is None or state.evaluated_revision is None + ): + reason = "unverified" + elif snapshot.inputs.profile is None: + reason = "insufficient_inputs" + elif state.evaluated_revision != state.input_revision: + reason = "inputs_changed" + elif evaluation.model_version != SCREENER_MODEL_VERSION: + reason = "model_changed" + elif evaluation.inputs_fingerprint != compute_inputs_fingerprint(snapshot.inputs): + reason = "evidence_or_time_changed" + return IpoCurrentEvaluation(snapshot, reason == "current", reason) + raise IpoScoringConflictError("IPO inputs changed during all three current-read attempts.") diff --git a/backend/ipo/scoring/state.py b/backend/ipo/scoring/state.py new file mode 100644 index 0000000..ada4654 --- /dev/null +++ b/backend/ipo/scoring/state.py @@ -0,0 +1,94 @@ +"""Detached IPO verification contracts shared by repository and scoring service.""" + +from __future__ import annotations + +import datetime as dt +from dataclasses import dataclass + +from backend.ipo.models import IpoEvaluationRecord +from backend.ipo.scoring.factor_derivation import IpoFactorInputs + + +class IpoScoringConflictError(RuntimeError): + """Retryable conflict after evidence changed during assembly/publication. + + Beginner note: + Callers may retry the operation later. They must never count this as a + successfully current evaluation or expose a half-published receipt. + """ + + code = "ipo_scoring_conflict" + retryable = True + + +@dataclass(frozen=True) +class IpoScoringStateRecord: + """Scalar state detached from SQLAlchemy's identity map. + + Beginner note: + Missing verification fields intentionally represent unverified history. + """ + + input_revision: int + current_score_id: int | None + evaluated_revision: int | None + last_verified_at: dt.datetime | None + + +@dataclass(frozen=True) +class IpoScoringSnapshot: + """One revision-validated evidence bundle and its selected historical pair. + + Beginner note: + Ratios derive from the detached bundle after the read transaction closes. + Current selection is read with the evidence, never in another transaction. + """ + + inputs: IpoFactorInputs + state: IpoScoringStateRecord + evaluation: IpoEvaluationRecord | None + + +@dataclass(frozen=True) +class IpoCurrentEvaluation: + """Current-read receipt retaining stale history for explicitly labeled display. + + Beginner note: + Only ``fresh`` permits an actionable verdict. The historical calculation + time remains on evaluation; last_verified_at describes a later check. + """ + + snapshot: IpoScoringSnapshot + fresh: bool + reason: str + + @property + def evaluation(self) -> IpoEvaluationRecord | None: + """Return selected history without implying freshness.""" + return self.snapshot.evaluation + + @property + def last_verified_at(self) -> dt.datetime | None: + """Return actual verification wall time independently of calculation.""" + return self.snapshot.state.last_verified_at + + +def normalize_scoring_time(value: dt.datetime) -> dt.datetime: + """Normalize an aware scoring clock to UTC; reject ambiguous naive times. + + Args: + value: Explicit evaluation/render instant. + + Returns: + The same instant in UTC, preserving the existing UTC near-close date rule. + + Raises: + ValueError: If value has no timezone offset. + + Beginner note: + An injected clock freezes business-time eligibility, not the real wall + time recorded when publication verifies the evidence. + """ + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("IPO scoring time must be timezone-aware.") + return value.astimezone(dt.UTC) diff --git a/backend/jobs/run_ipo_screener.py b/backend/jobs/run_ipo_screener.py index 8c6d764..c2ba6f4 100644 --- a/backend/jobs/run_ipo_screener.py +++ b/backend/jobs/run_ipo_screener.py @@ -49,6 +49,7 @@ SKIP, ) from backend.ipo.scoring.service import IpoRescoreOutcome, rescore_issue +from backend.ipo.scoring.state import IpoScoringConflictError from backend.ipo.sources.enrichment import collect_enrichment_signals from backend.jobs.scan_ipo_filings import IpoFilingJobOutcome, run_scan_ipo_filings from backend.observability import ( @@ -122,6 +123,8 @@ class IpoScreenerIssueOutcome: triggered_flags: tuple[str, ...] = () missing: tuple[str, ...] = () error_type: str | None = None + failure_code: str | None = None + retryable: bool = False @dataclass(frozen=True) @@ -531,6 +534,18 @@ def run_ipo_screener( outcome = _issue_outcome_from_rescore( rescorer(issue.id, session_factory=session_factory) ) + except IpoScoringConflictError as exc: + # Beginner note: all three attempts were already consumed by the + # service. Preserve sibling commits and expose a safe retry hint, + # never the exception message or a misleading successful verdict. + outcome = IpoScreenerIssueOutcome( + issue_id=issue.id, + company_name=issue.company_name, + status="failed", + error_type=type(exc).__name__, + failure_code=exc.code, + retryable=exc.retryable, + ) except Exception as exc: # noqa: BLE001 - per-issue isolation outcome = IpoScreenerIssueOutcome( issue_id=issue.id, diff --git a/backend/storage/ipo_repository.py b/backend/storage/ipo_repository.py index 3f7bb84..1550bdf 100644 --- a/backend/storage/ipo_repository.py +++ b/backend/storage/ipo_repository.py @@ -13,6 +13,8 @@ from __future__ import annotations import datetime as dt +import json +from collections.abc import Mapping from typing import Any, cast from sqlalchemy import func, select, update @@ -31,15 +33,145 @@ IpoManualPeerValuation, IpoRecommendation, IpoScore, + IpoScoringState, IpoSubscription, ) +def lock_ipo_scoring_state( + session: Session, issue_id: int, *, expected_revision: int | None = None +) -> bool: + """Acquire the shared issue serialization boundary in the caller transaction. + + Args: + session: Caller-owned write transaction; do not commit inside this helper. + issue_id: Issue whose state must be locked before subordinate input rows. + expected_revision: Optional publication CAS token captured with evidence. + + Returns: + Whether the state exists and the optional revision still matches. + + Beginner note: + A no-op UPDATE takes a database write lock until commit. All scoring + writers and downstream manual-baseline checks acquire this row first, + avoiding inverted lock order. It also starts SQLite's real transaction + before any nested savepoint. No manual-baseline authority is inferred + from this broad scoring-input revision. + """ + stmt = update(IpoScoringState).where(IpoScoringState.issue_id == issue_id) + if expected_revision is not None: + stmt = stmt.where(IpoScoringState.input_revision == expected_revision) + result = cast(CursorResult[Any], session.execute( + stmt.values(input_revision=IpoScoringState.input_revision) + .execution_options(synchronize_session=False) + )) + return result.rowcount == 1 + + +def advance_ipo_scoring_input_revision(session: Session, issue_id: int) -> None: + """Atomically invalidate verification before changing any scoring input. + + Args: + session: Caller-owned transaction retaining the lock through commit. + issue_id: Parent issue initialized with a state row on creation. + + Raises: + ValueError: If the issue lacks its required state row. + + Beginner note: + SQL addition avoids lost updates. Input and revision changes roll back + together. Multiple increments are safe: this is a monotonic mutation + token, not an edit count, and deliberately stays outside fingerprints. + """ + result = cast(CursorResult[Any], session.execute( + update(IpoScoringState).where(IpoScoringState.issue_id == issue_id) + .values(input_revision=IpoScoringState.input_revision + 1) + .execution_options(synchronize_session=False) + )) + if result.rowcount != 1 and get_ipo_issue(session, issue_id) is not None: + raise ValueError("IPO issue is missing its scoring state.") + + +def get_ipo_scoring_state_values( + session: Session, issue_id: int +) -> tuple[int, int | None, int | None, dt.datetime | None] | None: + """Read fresh scalar state without consulting cached ORM identity objects. + + Args: + session: Caller-owned read scope. + issue_id: Issue to inspect. + + Returns: + Revision, selected score, evaluated revision and UTC verification time, + or None when no state exists. + + Beginner note: + Repeating this SQL SELECT detects committed mutations between component + reads under READ COMMITTED; session.get could reuse an obsolete object. + """ + row = session.execute(select( + IpoScoringState.input_revision, IpoScoringState.current_score_id, + IpoScoringState.evaluated_revision, IpoScoringState.last_verified_at, + ).where(IpoScoringState.issue_id == issue_id)).one_or_none() + if row is None: + return None + verified = row[3] + if verified is not None and verified.tzinfo is None: + verified = verified.replace(tzinfo=dt.UTC) + return row[0], row[1], row[2], verified + + +def select_ipo_current_evaluation( + session: Session, issue_id: int, score_id: int, *, input_revision: int, verified_at: dt.datetime +) -> None: + """Publish a complete owned pair under the already-held state lock. + + Args: + session: Caller transaction holding lock_ipo_scoring_state. + issue_id: Owner of both the state and immutable score pair. + score_id: Historical score to select, including a reused older receipt. + input_revision: Evidence token verified by the caller's guarded calculation. + verified_at: Actual aware UTC wall time of successful verification. + + Raises: + ValueError: If the selected pair is incomplete, foreign, or revision changed. + + Beginner note: + The single-column FK cannot enforce ownership. Check both issue and score + here and on reads; no historical payload or calculation time is updated. + """ + if get_ipo_evaluation_rows(session, issue_id, score_id) is None: + raise ValueError("Current IPO evaluation must be complete and issue-owned.") + result = cast(CursorResult[Any], session.execute( + update(IpoScoringState).where( + IpoScoringState.issue_id == issue_id, + IpoScoringState.input_revision == input_revision, + ).values(current_score_id=score_id, evaluated_revision=input_revision, last_verified_at=verified_at) + .execution_options(synchronize_session=False) + )) + if result.rowcount != 1: + raise ValueError("IPO scoring revision changed before selection.") + + def insert_ipo_issue(session: Session, values: dict[str, Any]) -> IpoIssue: - """Stage one validated issue row and flush so its generated id is usable.""" + """Stage one issue and initialize its scoring state atomically. + + Args: + session: Caller-owned transaction, committed by the domain facade. + values: Validated issue columns. + + Returns: + The flushed issue with a generated identifier and revision-zero state. + + Beginner note: + State must exist from the same commit as the issue so every later input + writer has a shared serialization row. An issue without evidence has no + current score or verification timestamp.""" row = IpoIssue(**values) session.add(row) session.flush() + session.add(IpoScoringState(issue_id=row.id, input_revision=0)) + session.flush() return row @@ -78,7 +210,21 @@ def list_ipo_issue_rows(session: Session) -> list[IpoIssue]: def update_ipo_issue_row( session: Session, issue_id: int, values: dict[str, Any] ) -> IpoIssue | None: - """Apply supplied issue columns, refresh update time, and flush if present.""" + """Update an issue after atomically invalidating its scoring revision. + + Args: + session: Caller transaction retaining the state lock until commit. + issue_id: Parent issue to update. + values: Validated replacement columns. + + Returns: + The flushed issue, or None for a missing issue. + + Beginner note: + Ingestion calls this storage helper directly, so a facade-only hook would + miss scoring changes. Conservative invalidation is safe even when supplied + values are unchanged; rollback restores both evidence and revision.""" + advance_ipo_scoring_input_revision(session, issue_id) row = session.get(IpoIssue, issue_id) if row is None: return None @@ -102,7 +248,20 @@ def delete_ipo_issue_row(session: Session, issue_id: int) -> bool: def insert_ipo_document( session: Session, issue_id: int, values: dict[str, Any] ) -> IpoDocument: - """Stage one issue-owned source document and expose its generated id.""" + """Register a source under the issue's scoring serialization boundary. + + Args: + session: Caller-owned transaction. + issue_id: Owning issue. + values: Validated document columns. + + Returns: + Flushed document with its generated id. + + Beginner note: + Registered DRHP/RHP URLs enter the immutable scoring receipt, so source + registration invalidates verification even before a PDF is downloaded.""" + advance_ipo_scoring_input_revision(session, issue_id) row = IpoDocument(issue_id=issue_id, **values) session.add(row) session.flush() @@ -138,7 +297,22 @@ def get_ipo_document_by_url(session: Session, document_url: str) -> IpoDocument def update_ipo_document_values( session: Session, document: IpoDocument, values: dict[str, Any] ) -> IpoDocument: - """Mutate an already-owned document row and flush in the caller transaction.""" + """Update an owned document, invalidating source changes only. + + Args: + session: Caller-owned transaction. + document: Already-owned row to update. + values: Validated replacement columns. + + Returns: + The flushed document. + + Beginner note: + URL/type changes alter scoring provenance and acquire the state lock + before mutation. Cache-only metadata is display activity; the scorer does + not consume it, so downloading unchanged bytes cannot stale a score.""" + if {"document_url", "document_type"}.intersection(values): + advance_ipo_scoring_input_revision(session, document.issue_id) for name, value in values.items(): setattr(document, name, value) session.flush() @@ -166,7 +340,22 @@ def update_ipo_document_row( document_id: int, values: dict[str, Any], ) -> IpoDocument | None: - """Update a parent-scoped document or return ``None`` when ownership fails.""" + """Update a parent-scoped document under the source revision boundary. + + Args: + session: Caller-owned transaction. + issue_id: Required issue owner. + document_id: Document to update. + values: Validated columns; URL/type changes invalidate scoring. + + Returns: + Flushed document, or None if the ownership lookup fails. + + Beginner note: + Lock the state before changing provenance. Cache-only metadata does not + change the score's semantic inputs and therefore needs no revision bump.""" + if {"document_url", "document_type"}.intersection(values): + advance_ipo_scoring_input_revision(session, issue_id) row = get_ipo_document(session, issue_id, document_id) if row is None: return None @@ -211,7 +400,21 @@ def update_ipo_document_cache_if_source_matches( def delete_ipo_document_row(session: Session, issue_id: int, document_id: int) -> bool: - """Stage a parent-scoped metadata deletion without touching shared files.""" + """Delete registered provenance and invalidate verification atomically. + + Args: + session: Caller-owned transaction. + issue_id: Required issue owner. + document_id: Document to remove. + + Returns: + Whether a parent-scoped row was deleted. + + Beginner note: + Take the state write boundary before subordinate deletion. A failed + retention/foreign-key check rolls back the revision too; immutable score + receipts retain the URLs they originally used.""" + advance_ipo_scoring_input_revision(session, issue_id) row = get_ipo_document(session, issue_id, document_id) if row is None: return False @@ -290,11 +493,27 @@ def insert_ipo_manual_extraction( ) -> IpoManualExtraction: """Stage one complete immutable revision and all of its owned rows. + Args: + session: Caller-owned transaction. + issue_id: Parent issue. + header_values: Validated immutable extraction header. + period_values: Complete annual financial child rows. + peer_values: Approved peer valuation child rows. + + Returns: + Complete flushed extraction and eager children. + Beginner note: Header, periods, and peers are attached to one SQLAlchemy unit of work and flushed together. The caller's session context therefore either commits the complete revision or rolls every row back; a half-written form cannot exist. + + Scoring-state note: + The atomic revision increment acquires the shared state lock before input + writes or nested savepoints. It commits/rolls back with this evidence; + unchanged re-observation may reuse history after explicit verification. """ + advance_ipo_scoring_input_revision(session, issue_id) row = IpoManualExtraction(issue_id=issue_id, **header_values) row.periods = [ IpoManualFinancialPeriod(**values) for values in period_values @@ -376,7 +595,21 @@ def get_latest_ipo_manual_extraction( def insert_ipo_subscription( session: Session, issue_id: int, values: dict[str, Any] ) -> IpoSubscription: - """Stage one timestamped demand snapshot under its parent issue.""" + """Append demand evidence after locking and advancing scoring state. + + Args: + session: Caller-owned transaction. + issue_id: Owning issue. + values: Validated subscription columns. + + Returns: + The flushed demand snapshot. + + Beginner note: + A latest-demand change may alter factors or near-close caution flags. + Advancing before insertion makes publication either see the complete new + evidence or fail its captured-revision check.""" + advance_ipo_scoring_input_revision(session, issue_id) row = IpoSubscription(issue_id=issue_id, **values) session.add(row) session.flush() @@ -412,7 +645,21 @@ def update_ipo_subscription_row( subscription_id: int, values: dict[str, Any], ) -> IpoSubscription | None: - """Replace selected fields on a parent-scoped demand snapshot and flush.""" + """Update demand evidence and its scoring revision in one transaction. + + Args: + session: Caller-owned transaction. + issue_id: Required issue owner. + subscription_id: Snapshot to update. + values: Validated replacement columns. + + Returns: + Flushed snapshot or None for an ownership mismatch. + + Beginner note: + Even editing an older observation conservatively invalidates verification. + State locks precede row writes, and rollback restores both changes.""" + advance_ipo_scoring_input_revision(session, issue_id) row = get_ipo_subscription(session, issue_id, subscription_id) if row is None: return None @@ -425,7 +672,20 @@ def update_ipo_subscription_row( def delete_ipo_subscription_row( session: Session, issue_id: int, subscription_id: int ) -> bool: - """Stage deletion of one issue-owned snapshot and remain idempotent.""" + """Delete demand evidence after acquiring the shared state boundary. + + Args: + session: Caller-owned transaction. + issue_id: Required issue owner. + subscription_id: Snapshot to remove. + + Returns: + Whether a parent-scoped snapshot existed and was deleted. + + Beginner note: + Removing the latest row makes an older capture authoritative. The revision + must advance even though this operation inserts no replacement evidence.""" + advance_ipo_scoring_input_revision(session, issue_id) row = get_ipo_subscription(session, issue_id, subscription_id) if row is None: return False @@ -645,6 +905,14 @@ def insert_ipo_enrichment_signals( ) -> list[IpoEnrichmentSignal]: """Stage one enrichment batch for an issue as a single unit of work. + Args: + session: Caller-owned transaction. + issue_id: Parent issue. + values_list: Validated observation columns for a compatibility batch. + + Returns: + All flushed observations, without committing the caller transaction. + A SerpAPI collection run produces several signal types at one capture instant; inserting them together keeps a partially-persisted batch from masquerading as a complete observation set. @@ -653,7 +921,13 @@ def insert_ipo_enrichment_signals( This older batch helper remains for compatibility. New collection code uses semantic upsert so repeated observations refresh rather than accumulate. + + Scoring-state note: + The atomic revision increment acquires the shared state lock before input + writes or nested savepoints. It commits/rolls back with this evidence; + unchanged re-observation may reuse history after explicit verification. """ + advance_ipo_scoring_input_revision(session, issue_id) rows = [IpoEnrichmentSignal(issue_id=issue_id, **values) for values in values_list] session.add_all(rows) session.flush() @@ -686,11 +960,25 @@ def upsert_ipo_enrichment_signal( ) -> IpoEnrichmentSignal: """Preserve first-seen identity and refresh last-seen on identical evidence. + Args: + session: Caller-owned transaction. + issue_id: Parent issue. + values: Validated observation with semantic identity and aware timestamps. + + Returns: + The inserted or refreshed semantic observation. + Beginner note: The preflight read is an optimization, not a concurrency guarantee. When two collectors race, the unique semantic index selects one row and the losing savepoint reloads it before refreshing freshness timestamps. + + Scoring-state note: + The atomic revision increment acquires the shared state lock before input + writes or nested savepoints. It commits/rolls back with this evidence; + unchanged re-observation may reuse history after explicit verification. """ + advance_ipo_scoring_input_revision(session, issue_id) semantic_hash = str(values["semantic_hash"]) signal_type = str(values["signal_type"]) existing = _get_ipo_enrichment_signal_by_semantic_hash( @@ -753,6 +1041,79 @@ def list_ipo_enrichment_signal_rows( return list(session.scalars(stmt)) +def _canonical_evaluation_value(name: str, value: Any) -> Any: + """Represent one receipt field for strict, non-mutating comparison. + + Args: + name: Persisted column name from a freshly derived candidate payload. + value: Candidate or stored column value; no caller text is normalized away. + + Returns: + A stable JSON representation for JSON columns, otherwise the scalar value. + + Raises: + TypeError: If a JSON field contains a non-serializable value. + ValueError: If a JSON field contains a circular container. + + Beginner note: + Sources and missing-factor labels are sets; their order and duplicates do + not change meaning. Other receipt lists retain their full ordered content. + Canonical JSON sorts mapping keys and distinguishes booleans from numbers, + so Python's ``True == 1`` cannot conceal a changed JSON receipt. Scalar + Decimal comparisons ignore harmless numeric scale without losing precision. + """ + # Malformed set elements remain uncanonicalized and cannot equal a valid + # candidate. Never repair or rewrite historical evidence during this check. + if ( + name in {"source_documents_json", "missing_data_json"} + and isinstance(value, list) + and all(isinstance(item, str) for item in value) + ): + value = sorted(set(value)) + if name.endswith("_json"): + return json.dumps(value, sort_keys=True, separators=(",", ":")) + return value + + +def ipo_evaluation_payload_matches( + score: IpoScore, + recommendation: IpoRecommendation, + *, + score_values: Mapping[str, Any], + recommendation_values: Mapping[str, Any], +) -> bool: + """Check every derived persisted field before trusting a historical pair. + + Args: + score: Historical score returned by the semantic uniqueness boundary. + recommendation: Its complete paired recommendation. + score_values: Full candidate factors, total, contribution/breakdown/reason + receipts, missing labels, model and fingerprint derived by the service. + recommendation_values: Full candidate verdict, confidence, reasons, missing + labels, cautions and source provenance derived by the service. + + Returns: + True only when all candidate payload fields agree with persisted content. + + Raises: + TypeError: If candidate or stored JSON cannot be serialized. + ValueError: If candidate or stored JSON contains a circular container. + + Beginner note: + The legacy API lets callers assert a model/fingerprint, so uniqueness is + not proof that its scorecard was derived from those inputs. Publication + calls this under the state lock and rejects a mismatch before selecting + or verifying the pair. IDs and calculation/verification times are absent + from candidate payloads and must never make legitimate A-B-A reuse fail. + This helper reads loaded values only and never edits historical receipts. + """ + return all( + _canonical_evaluation_value(name, getattr(row, name)) == _canonical_evaluation_value(name, value) + for row, values in ((score, score_values), (recommendation, recommendation_values)) + for name, value in values.items() + ) + + def insert_ipo_evaluation( session: Session, issue_id: int, diff --git a/backend/storage/models.py b/backend/storage/models.py index 66af365..4301abd 100644 --- a/backend/storage/models.py +++ b/backend/storage/models.py @@ -1341,6 +1341,31 @@ class IpoSubscription(Base): issue: Mapped[IpoIssue] = relationship(back_populates="subscriptions") +class IpoScoringState(Base): + """Mutable selection and verification metadata beside immutable history. + + Beginner note: + The input revision is an invalidation token, not a count of manual edits. + An unchanged evidence refresh may reuse an old score but must verify the + new revision. A score pointer alone never certifies current evidence. + """ + + __tablename__ = "ipo_scoring_state" + __table_args__ = ( + CheckConstraint("input_revision >= 0", name="ck_ipo_scoring_state_input_revision"), + CheckConstraint( + "evaluated_revision >= 0 AND evaluated_revision <= input_revision", + name="ck_ipo_scoring_state_evaluated_revision", + ), + ) + + issue_id: Mapped[int] = mapped_column(ForeignKey("ipo_issues.id", ondelete="CASCADE"), primary_key=True) + input_revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default="0") + current_score_id: Mapped[int | None] = mapped_column(ForeignKey("ipo_scores.id", ondelete="SET NULL")) + evaluated_revision: Mapped[int | None] = mapped_column(Integer) + last_verified_at: Mapped[dt.datetime | None] = mapped_column(DateTime(timezone=True)) + + class IpoScore(Base): """Persist one immutable seven-factor calculation and its audit breakdown. diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 23317eb..ce60128 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -110,3 +110,5 @@ key decision changes, update its LLD (and the HLD if the change is system-wide) the same PR. - [VALID-005 forward-return integrity and retry scheduling](valid-005-forward-return-integrity.md) — immutable receipts, short transactions, fair stock and benchmark retries. + +- **[ipo-013-current-evaluation-state.md](ipo-013-current-evaluation-state.md)** — immutable IPO history, mutable verified selection, revision/CAS publication and semantic/time freshness. diff --git a/docs/architecture/components/ipo-screener.md b/docs/architecture/components/ipo-screener.md index f860401..16f4af4 100644 --- a/docs/architecture/components/ipo-screener.md +++ b/docs/architecture/components/ipo-screener.md @@ -106,7 +106,7 @@ network or model stages. All persistence still routes through `backend/storage`. | `backend/ipo/financials/ratio_engine.py` | Pure Decimal formulas, typed status receipts, reconciliation, source/price snapshot. | stdlib, `manual_extraction` | | `backend/ipo/scoring/factor_derivation.py` | Pure seven-factor derivation plus typed, negation-aware debt-purpose evidence. | `models`, `ratio_engine`, `manual_extraction` | | `backend/ipo/scoring/caution_flags.py` | Seven fixed-order hard cautions over typed evidence authority. | `models`, `factor_derivation`, `ratio_engine` | -| `backend/ipo/scoring/service.py` | One-transaction input snapshot, semantic fingerprint, idempotent evaluation orchestration. | `repository`, pure scoring modules | +| `backend/ipo/scoring/service.py` | Revision-validated input snapshot, semantic fingerprint, conditional current selection. | `repository`, pure scoring modules | | `backend/ipo/documents/table_extractor.py` | Parent-owned PDF facade, worker supervision, result-budget validation, typed parse receipts. | stdlib, `backend/ipo_pdf_worker.py` | | `backend/ipo_pdf_worker.py` | Dependency-light spawn entrypoint; applies Linux address-space policy before lazy pdfplumber import and emits primitive JSON only. | stdlib, lazy `pdfplumber` | | `backend/ipo/documents/section_classifier.py` | Page/span-preserving heading ownership and safe chunks. | `table_extractor` | @@ -131,7 +131,7 @@ no IPO module imports Streamlit, and network clients are allowed only under | `score_ipo(IpoScoreInput) -> IpoScoreResult` | Applies the fixed weights; missing factors contribute zero and are never renormalized. | | `build_recommendation(IpoScoreResult) -> IpoRecommendationResult` | Maps a score to the binary verdict + confidence; `.to_dict()` includes all seven typed breakdown rows. | | `evaluate_issue(issue_id, IpoScoreInput)` | Computes and atomically persists one immutable score/verdict pair; a semantic uniqueness race returns the winning pair. | -| `load_ipo_factor_inputs_snapshot(issue_id, as_of=...)` | Reads issue/profile/subscription/enrichment in one transaction, then derives ratios from that exact detached snapshot. | +| `load_ipo_factor_inputs_snapshot(issue_id, as_of=...)` | Validates scalar revision around all input reads, then derives ratios from the detached bundle. | | `extract_document_pages(path, budget=...)` | Compatible PDF facade returning a typed success or review-required receipt from a killable child. | | `collect_enrichment_signals(issue_id, ...)` | Uses persisted issuer/price identity, quarantines each result, and semantically upserts advisory observations. | | `propose_extraction(issue_id, document_id, force_extract=False)` | Produces a citation-bound review proposal or a stable skip/failure receipt; never scoring evidence. | @@ -332,3 +332,20 @@ design doc: Any extension must preserve URL safety, never invent missing evidence, and route all SQL through `backend/storage`. + +## Current evaluation and verification (IPO-013) + +[IPO-013](../ipo-013-current-evaluation-state.md) separates immutable receipts from +`ipo_scoring_state`: one monotonic scoring-input revision and mutable current +pointer, evaluated revision and verification time per issue. All scoring writers +lock that state before subordinate changes. Snapshot assembly validates fresh SQL +scalar state before/after all component reads; publication conditionally locks the +same captured revision. Both conflict sites share three total attempts. + +`get_latest_evaluation`, `list_evaluations` and `get_latest_recommendation` are +historical APIs. `get_current_evaluation` checks ownership, complete pair, verified +revision, current model and semantic/time fingerprint. Dashboard renders check +freshness every rerun; stale history cannot enter actionable recommendation +sections or screener ratings. Calculation time, verification time and display +activity are distinct. Migration `20260921ipo013` follows `20260909valid005` and +backfills only unverified display pointers, preserving every historical receipt. diff --git a/docs/architecture/high-level-design.md b/docs/architecture/high-level-design.md index 39536b0..faf49c5 100644 --- a/docs/architecture/high-level-design.md +++ b/docs/architecture/high-level-design.md @@ -24,7 +24,9 @@ job — is recorded to a scan-history database. Access is gated behind Google SS with an email allowlist for the interactive Streamlit surface. The backend also inventories official SEBI DRHP, RHP, and final-offer listings -and stores immutable IPO score/recommendation history. The explicit IPO screener +and stores immutable IPO score/recommendation history with a separate verified +current selection ([IPO-013](ipo-013-current-evaluation-state.md)). +Input revisions guard publication; current views check model and time freshness. The explicit IPO screener job composes inventory, verified DRHP/RHP download, optional advisory enrichment, optional citation-bound extraction proposals, and deterministic re-scoring. Administrators may enter evidence manually or approve a proposal; only approved diff --git a/docs/architecture/ipo-007-dashboard.md b/docs/architecture/ipo-007-dashboard.md index 46b1274..952454c 100644 --- a/docs/architecture/ipo-007-dashboard.md +++ b/docs/architecture/ipo-007-dashboard.md @@ -8,14 +8,16 @@ reads, and a thin page (`ui/ipo_page.py`) renders whatever the builder returned. No network call and no scoring happen during render; the compute pass is the IPO-008 job or the page's explicit re-score action. -`build_dashboard_snapshot` denormalizes each issue's stored state — latest +`build_dashboard_snapshot` denormalizes each issue's stored state — current selected evaluation (with its contribution receipt), manual-profile presence, cached document counts, pending proposal count — into frozen `IpoDashboardRow` values. Pure selectors implement the seven spec sections: Available filings, Open IPOs, Upcoming IPOs (RHP stage), DRHP watchlist, Recommended, Not Recommended, and the Missing data queue (no verified profile, no downloaded prospectus, a factor the verdict flagged missing, or a proposal awaiting -review). `top_positive_and_risk_reasons` ranks stored contributions against +review, or stale/unverified evaluation). Recommended and Not Recommended +sections require semantic freshness; historical verdicts remain in All. +`top_positive_and_risk_reasons` ranks stored contributions against `PDF_WEIGHTS` (>=75% of weight is a headline strength, <=35% a headline risk); missing factors are excluded because "could not check" and "checked and weak" are deliberately different messages. @@ -40,7 +42,10 @@ and weak" are deliberately different messages. (hiding is UX; the app dispatch capability check is the boundary). It runs the same `rescore_issue` service the job uses — repository work only — counts outcomes without letting one failure abort the rest, records an - audit event, and invalidates the five-minute snapshot cache. + audit event, and rebuilds the snapshot. There is no actionable snapshot cache: + each rerun checks input revision, model and time-derived semantics at one UTC + instant. Calculation and verification time are displayed separately from + overall display activity. ## Testing @@ -49,14 +54,19 @@ missing-data queue rules, strength/risk selection, and snapshot denormalization over monkeypatched repositories. `tests/test_app_ipo_page.py` smoke-tests the renderer against a fake ``st`` with every repository seam stubbed (proving render purity), plus the label -map, filter semantics, spec column contract, and the re-score audit/cache +map, filter semantics, spec column contract, and the re-score audit/refresh path. `tests/test_app_orchestration.py` pins the navigation entry, the re-export identity, and the keyword-only capability boundary. > PR #108 hardening: scored and unscored rows expose registered DRHP/RHP source > documents; `last_updated` is the newest relevant issue/document/profile/ > proposal/subscription/enrichment/evaluation time; `evaluation_stale` routes -> newer evidence back to the review queue. Positives sort by awarded points, +> unverified evidence or changed model/time eligibility back to the review queue. Positives sort by awarded points, > risks by lost points with factor-order ties. Every scored expander renders all > seven typed breakdown rows, and untrusted text is escaped before any > Markdown-capable Streamlit label/caption. + +The transaction and freshness authority is [IPO-013](ipo-013-current-evaluation-state.md). +Cache downloads and unapproved proposal changes may update display activity but +do not alone stale a verified score. A still-fresh GMP observation can require +reverification without creating a new historical calculation. diff --git a/docs/architecture/ipo-008-screener-orchestration.md b/docs/architecture/ipo-008-screener-orchestration.md index f6a2086..531e166 100644 --- a/docs/architecture/ipo-008-screener-orchestration.md +++ b/docs/architecture/ipo-008-screener-orchestration.md @@ -18,24 +18,21 @@ and writes nothing — missing data never becomes a fabricated score. ## Inputs fingerprint (the idempotency anchor) -Before persisting, the service computes a SHA-256 over exactly what scoring -consumed: the three rule versions (`ipo-006-v1`, factor and flag versions), -the extraction id + source SHA-256, the issue's updated-at/status/price -band, the newest subscription snapshot identity, every enrichment signal id, -and two *time-derived* facts — the set of GMP signals still inside the -staleness window and whether the issue is inside its near-close demand -window. Hashing derived facts instead of the clock keeps re-runs no-ops -until the passage of time would actually change a factor or flag. When the -newest stored evaluation carries the same model version and fingerprint the -service reports `skipped_unchanged`; the fingerprint is stored on -`ipo_scores.inputs_fingerprint` (legacy ipo-001-v1 rows keep `NULL`). +The service hashes semantic evidence, source digests, canonical registered source +URLs, ratio/authority/rule versions, subscription facts and usable enrichment. +The current scorer version is `ipo-013-v3`; database ids, mutation revisions and +raw wall time remain excluded. GMP eligibility includes the exact five-day +threshold and expires strictly after it. Near-close uses the existing UTC date. -The hardened implementation supersedes the volatile identities named above: -it hashes rule/ratio/authority versions, source SHA and normalized approved -values, ratio statuses/results, subscription facts, usable semantic enrichment -facts, typed debt-purpose evidence, GMP freshness, and the near-close state. -Database row ids are excluded, and a partial unique index closes concurrent -check/insert races. +A revision-validated detached snapshot feeds calculation. Conditional publication +locks the issue state before inserting/reusing the unique complete immutable pair. +A -> B -> A selects historical A again. Unchanged content reports +`skipped_unchanged` while updating evaluated revision and actual verification time. +Historical calculation time and receipt bytes never change. Reads never certify +freshness or insert history. Snapshot and publication conflicts share at most three +total attempts, after which the job reports a typed retryable per-issue failure. +Production rechecks time eligibility before publication; explicit aware `as_of` +freezes the business clock. See [IPO-013](ipo-013-current-evaluation-state.md). ## Failure and configuration semantics diff --git a/docs/architecture/ipo-013-current-evaluation-state.md b/docs/architecture/ipo-013-current-evaluation-state.md new file mode 100644 index 0000000..fc0a1ea --- /dev/null +++ b/docs/architecture/ipo-013-current-evaluation-state.md @@ -0,0 +1,64 @@ +# IPO-013: current evaluation and semantic freshness + +Status: Accepted for the approved modernization Task 8, 2026-09-22. + +## Decision + +Keep scores and recommendations immutable. Add one `ipo_scoring_state` row per +issue containing a monotonic `input_revision`, nullable `current_score_id`, +`evaluated_revision`, and `last_verified_at`. Historical newest and current are +different: A -> B -> A reselects the original A receipt without changing its +calculation time or duplicating history. Migration `20260921ipo013` follows +`20260909valid005`; backfill selects the newest complete pair by calculation time +then score id, but never certifies legacy history as verified. + +## Transactions and shared review boundary + +Every actual scoring-input writer atomically increments the state revision in +the caller's transaction before changing input rows. The state row is always +locked before subordinate rows. `lock_ipo_scoring_state(session, issue_id, +expected_revision=None)` is the shared serialization interface for downstream +manual-review work. Its no-op SQL UPDATE holds the lock until caller commit. +The downstream manual-only baseline must remain distinct from input_revision: +subscription and GMP refreshes must not pretend to be manual edits. + +Read scalar state before and after detaching every input and eager child. +SQL column reads bypass the identity map; a changed token discards the bundle, +including A -> B -> A mutations during assembly. Read-time selection metadata +belongs to that same bundle. Derive factors outside the read transaction. +Publication starts a new transaction and conditionally locks the captured +revision before inserting/reusing a complete owned score pair. The pointer, +verified revision and verification time commit together. This real outer UPDATE +also encloses SQLite savepoints, so late failure rolls back inserted history. +There are at most three total attempts across snapshot and publication conflicts. +Unrelated database/validation failures retain their original meaning. + +## Freshness and compatibility + +Current means a complete issue-owned pair, verified matching revision, present +verification time, and model/fingerprint matching current inputs at the read +clock. Canonical source URLs enter the new versioned fingerprint. Raw time and +the mutation token never enter semantic identity. GMP is valid at exactly five +days and expires immediately after; near-close retains its existing UTC date +rule. Explicit aware `as_of` freezes evaluation time; production checks the +time-derived fingerprint again before publication. Verification time is actual +UTC wall time and remains independent of the immutable calculation time. + +Current dashboard snapshots rebuild each render with one aware UTC instant. +Display activity (proposals/cache downloads) may change last_updated without +invalidating scoring. Stale receipts remain visible as history but cannot enter +actionable recommendation sections or screener ratings. Missing manual evidence +does not fabricate a score or certify an old pointer. Legacy evaluation APIs +remain historical and cannot certify arbitrary caller-supplied scorecards. + +## Alternatives and validation + +Timestamp comparisons miss semantic reuse, clock expiry and concurrent commits. +Rewriting receipts destroys audit history; deleting uniqueness creates duplicates. +Fixed UI cache TTLs cannot prove freshness after another session writes. +The selected design uses existing transaction scopes and unique/savepoint logic. + +Regression coverage must include semantic return to A, refresh without duplicate +history, all storage writers and rollback, snapshot/publication interleavings, +three-attempt exhaustion, ownership, migration retention, clock boundaries, +uncached actionable reads, and deterministic same-input concurrency. diff --git a/migrations/versions/20260921ipo013_current_scoring_state.py b/migrations/versions/20260921ipo013_current_scoring_state.py new file mode 100644 index 0000000..ce7a447 --- /dev/null +++ b/migrations/versions/20260921ipo013_current_scoring_state.py @@ -0,0 +1,51 @@ +"""Separate current verification from immutable IPO scoring history. + +Revision ID: 20260921ipo013 +Revises: 20260909valid005 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision = "20260921ipo013" +down_revision = "20260909valid005" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Backfill display pointers without granting legacy receipts freshness. + + Beginner note: + Joining the recommendation excludes orphan scores. Equal calculation + times use the greatest id as a deterministic tie break. No historical + receipt or timestamp is rewritten; verification remains explicitly NULL. + """ + op.create_table( + "ipo_scoring_state", + sa.Column("issue_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), + sa.ForeignKey("ipo_issues.id", ondelete="CASCADE"), primary_key=True), + sa.Column("input_revision", sa.Integer(), server_default="0", nullable=False), + sa.Column("current_score_id", sa.BigInteger().with_variant(sa.Integer(), "sqlite"), + sa.ForeignKey("ipo_scores.id", ondelete="SET NULL"), nullable=True), + sa.Column("evaluated_revision", sa.Integer(), nullable=True), + sa.Column("last_verified_at", sa.DateTime(timezone=True), nullable=True), + sa.CheckConstraint("input_revision >= 0", name="ck_ipo_scoring_state_input_revision"), + sa.CheckConstraint("evaluated_revision >= 0 AND evaluated_revision <= input_revision", + name="ck_ipo_scoring_state_evaluated_revision"), + ) + op.execute(sa.text( + "INSERT INTO ipo_scoring_state (issue_id, input_revision, current_score_id) " + "SELECT i.id, 0, (SELECT s.id FROM ipo_scores s " + "JOIN ipo_recommendations r ON r.score_id = s.id WHERE s.issue_id = i.id " + "ORDER BY s.scored_at DESC, s.id DESC LIMIT 1) FROM ipo_issues i" + )) + + +def downgrade() -> None: + """Remove mutable selection only; every historical pair is retained.""" + op.drop_table("ipo_scoring_state") diff --git a/screeners/ipo_screener.py b/screeners/ipo_screener.py index fcf05bf..ecca1e0 100644 --- a/screeners/ipo_screener.py +++ b/screeners/ipo_screener.py @@ -294,9 +294,12 @@ def _result_row( if failed: triggered_rules.append("scoring_failed") - if row.recommendation_type: + if row.evaluation_stale: + triggered_rules.append("evaluation_stale") + if row.recommendation_type and not row.evaluation_stale: triggered_rules.append(f"verdict:{row.recommendation_type}") - triggered_rules.extend(row.triggered_flags) + if not row.evaluation_stale: + triggered_rules.extend(row.triggered_flags) if row.missing_data: triggered_rules.append("missing_data") if not triggered_rules: @@ -322,7 +325,8 @@ def _result_row( return { "symbol": f"IPO:{row.issue_id}", - "rating": row.recommendation, + "rating": row.recommendation if not row.evaluation_stale else None, + "historical_recommendation": row.recommendation, # Deliberately null. A forward return is "the price N sessions after # the signal date", which an IPO issue has no series for: the VALID-002 # job selects every result carrying a signal_date, could never resolve @@ -330,13 +334,14 @@ def _result_row( # these rows as PENDING forever, consuming its batch budget. The # evaluation date is still reported, in its own column. "signal_date": None, - "scored_on": row.last_updated.date() if row.last_updated else None, + "scored_on": row.calculated_at.date() if row.calculated_at else None, "close": row.score, "reason": row.reasons[0] if row.reasons else "No evaluation yet.", "company_name": row.company_name, "issue_status": row.issue_status.value, "ipo_score": row.score, - "recommendation_type": row.recommendation_type, + "recommendation_type": row.recommendation_type if not row.evaluation_stale else None, + "historical_recommendation_type": row.recommendation_type, "confidence": row.confidence, "top_positives": "; ".join(row.top_positives), "top_risks": "; ".join((*row.triggered_flags, *row.top_risks)), diff --git a/tests/test_app_ipo_page.py b/tests/test_app_ipo_page.py index 28c4d22..7f98eca 100644 --- a/tests/test_app_ipo_page.py +++ b/tests/test_app_ipo_page.py @@ -134,6 +134,8 @@ def test_rows_frame_carries_every_spec_column() -> None: "Source documents", "Last updated", "Evaluation stale", + "Calculated at", + "Last verified at", ] record = frame.iloc[0] assert record["Company"] == "Example Ltd" @@ -388,7 +390,7 @@ def _record_audit(**kwargs: Any) -> bool: ipo_page._render_ipo_page(can_rescore=True, user_email="admin@example.com") assert rescored == [1, 2] - assert loader.cleared == 1 + assert loader.cleared == 0 assert audits[0]["user_email"] == "admin@example.com" assert audits[0]["metadata"]["skipped_unchanged"] == 2 assert any("Re-score complete" in message for message in fake_st.successes) @@ -416,3 +418,17 @@ def _rescore(issue_id: int, **_kwargs: Any) -> IpoRescoreOutcome: assert any("1 failed" in message for message in fake_st.successes) assert any("1 evaluated" in message for message in fake_st.successes) + +def test_snapshot_loader_checks_current_state_every_call(monkeypatch) -> None: + """Beginner note: a global TTL hid other sessions' writes and clock expiry.""" + from ui import ipo_page + + outputs = iter([ + IpoDashboardSnapshot(generated_at=_SCORED_AT, rows=()), + IpoDashboardSnapshot(generated_at=_SCORED_AT + dt.timedelta(seconds=1), rows=()), + ]) + monkeypatch.setattr(ipo_page, "build_dashboard_snapshot", lambda: next(outputs)) + first = ipo_page._load_snapshot() + second = ipo_page._load_snapshot() + assert first.generated_at == _SCORED_AT + assert second.generated_at == _SCORED_AT + dt.timedelta(seconds=1) diff --git a/tests/test_ipo_auto_approval.py b/tests/test_ipo_auto_approval.py index 5733733..e364d9f 100644 --- a/tests/test_ipo_auto_approval.py +++ b/tests/test_ipo_auto_approval.py @@ -11,7 +11,7 @@ from __future__ import annotations from types import SimpleNamespace -from typing import Any +from typing import Any, cast from backend.ipo.agents.auto_approval import ( AUTOMATION_ACTOR_EMAIL, @@ -23,6 +23,7 @@ IpoValidationError, ) from backend.observability import EVENT_IPO_PROPOSAL_AUTO_APPROVED +from backend.storage import SessionFactory def _proposal(proposal_id: int, confidence: Confidence) -> Any: @@ -53,7 +54,7 @@ def _must_not_approve(*_args: Any, **_kwargs: Any) -> Any: raise AssertionError("auto-approval must not run when disabled") outcome = auto_approve_ready_proposals( - enabled=False, approver=_must_not_approve, session_factory=object + enabled=False, approver=_must_not_approve, session_factory=cast(SessionFactory, object) ) assert outcome.disabled is True @@ -87,7 +88,7 @@ def _approver(proposal_id: int, **_kwargs: Any) -> Any: enabled=True, approver=_approver, audit_recorder=lambda **_kwargs: True, - session_factory=object, + session_factory=cast(SessionFactory, object), ) assert approved_ids == [1] @@ -116,7 +117,7 @@ def _record_audit(**kwargs: Any) -> bool: enabled=True, approver=_approver, audit_recorder=_record_audit, - session_factory=object, + session_factory=cast(SessionFactory, object), ) assert seen["reviewed_by_email"] == AUTOMATION_ACTOR_EMAIL @@ -145,7 +146,7 @@ def _approver(proposal_id: int, **_kwargs: Any) -> Any: enabled=True, approver=_approver, audit_recorder=lambda **_kwargs: True, - session_factory=object, + session_factory=cast(SessionFactory, object), ) assert approved_ids == [2] @@ -161,7 +162,7 @@ def test_environment_switch_drives_the_default(monkeypatch) -> None: monkeypatch.setattr( auto_approval, "get_ipo_auto_approve_high_confidence", lambda: False ) - assert auto_approve_ready_proposals(session_factory=object).disabled is True + assert auto_approve_ready_proposals(session_factory=cast(SessionFactory, object)).disabled is True monkeypatch.setattr( auto_approval, "get_ipo_auto_approve_high_confidence", lambda: True @@ -169,7 +170,7 @@ def test_environment_switch_drives_the_default(monkeypatch) -> None: outcome = auto_approve_ready_proposals( approver=lambda proposal_id, **_kwargs: SimpleNamespace(id=proposal_id), audit_recorder=lambda **_kwargs: True, - session_factory=object, + session_factory=cast(SessionFactory, object), ) assert outcome.disabled is False assert outcome.approved == (1,) diff --git a/tests/test_ipo_dashboard_builder.py b/tests/test_ipo_dashboard_builder.py index f9fa3d5..ff2d60a 100644 --- a/tests/test_ipo_dashboard_builder.py +++ b/tests/test_ipo_dashboard_builder.py @@ -13,7 +13,7 @@ import datetime as dt from decimal import Decimal from types import SimpleNamespace -from typing import Any +from typing import Any, cast from backend.ipo import dashboard from backend.ipo.dashboard import ( @@ -39,6 +39,7 @@ IpoStatus, Recommendation, ) +from backend.storage import SessionFactory _SCORED_AT = dt.datetime(2026, 7, 13, 9, 0, tzinfo=dt.UTC) @@ -193,6 +194,16 @@ def test_missing_data_queue_catches_every_evidence_gap() -> None: assert [row.issue_id for row in queued] == [2, 3, 4, 5] +def test_stale_receipts_never_enter_actionable_sections() -> None: + """Beginner note: the old sections treated stale positive/negative history as current.""" + rows = (_row(evaluation_stale=True), _row(issue_id=2, evaluation_stale=True, recommendation="Not Recommended")) + snapshot = IpoDashboardSnapshot(generated_at=_SCORED_AT, rows=rows) + assert section_recommended(snapshot) == () + assert section_not_recommended(snapshot) == () + assert section_available_filings(snapshot) == rows + assert section_missing_data_queue(snapshot) == rows + + def test_build_snapshot_denormalizes_stored_state_per_issue(monkeypatch) -> None: """The builder reads repositories only and flattens them into rows.""" issue_updated = _SCORED_AT - dt.timedelta(days=2) @@ -246,18 +257,6 @@ def test_build_snapshot_denormalizes_stored_state_per_issue(monkeypatch) -> None "list_documents", lambda issue_id, **_kwargs: documents[issue_id], ) - monkeypatch.setattr( - dashboard, - "get_latest_manual_profile", - lambda issue_id, **_kwargs: ( - SimpleNamespace( - source_document_url="https://www.sebi.gov.in/scored-rhp", - submitted_at=issue_updated, - ) - if issue_id == 1 - else None - ), - ) monkeypatch.setattr( dashboard, "list_extraction_proposals", @@ -284,12 +283,18 @@ def test_build_snapshot_denormalizes_stored_state_per_issue(monkeypatch) -> None lambda *_args, **_kwargs: [], ) monkeypatch.setattr( - dashboard, - "get_latest_evaluation", - lambda issue_id, **_kwargs: evaluation if issue_id == 1 else None, + dashboard, "get_current_evaluation", + lambda issue_id, **_kwargs: SimpleNamespace( + evaluation=evaluation if issue_id == 1 else None, + fresh=issue_id == 1, last_verified_at=_SCORED_AT if issue_id == 1 else None, + snapshot=SimpleNamespace(inputs=SimpleNamespace( + issue=issues[issue_id - 1], source_documents=tuple(d.document_url for d in documents[issue_id]), + profile=SimpleNamespace(submitted_at=issue_updated) if issue_id == 1 else None, + )), + ), ) - snapshot = build_dashboard_snapshot(now=_SCORED_AT, session_factory=object) + snapshot = build_dashboard_snapshot(now=_SCORED_AT, session_factory=cast(SessionFactory, object)) assert snapshot.generated_at == _SCORED_AT scored, fresh = snapshot.rows @@ -310,7 +315,7 @@ def test_build_snapshot_denormalizes_stored_state_per_issue(monkeypatch) -> None def test_newer_evidence_marks_the_displayed_evaluation_stale( monkeypatch, ) -> None: - """Fresh evidence after scored_at routes the issue back to the review queue.""" + """A changed input revision routes the issue back to the review queue.""" newer = _SCORED_AT + dt.timedelta(hours=1) issue = SimpleNamespace( id=1, @@ -329,14 +334,6 @@ def test_newer_evidence_marks_the_displayed_evaluation_stale( monkeypatch.setattr( dashboard, "list_documents", lambda *_args, **_kwargs: [document] ) - monkeypatch.setattr( - dashboard, - "get_latest_manual_profile", - lambda *_args, **_kwargs: SimpleNamespace( - source_document_url=document.document_url, - submitted_at=_SCORED_AT, - ), - ) monkeypatch.setattr( dashboard, "list_extraction_proposals", lambda **_kwargs: [] ) @@ -351,16 +348,20 @@ def test_newer_evidence_marks_the_displayed_evaluation_stale( dashboard, "list_enrichment_signals", lambda *_args, **_kwargs: [] ) monkeypatch.setattr( - dashboard, - "get_latest_evaluation", - lambda *_args, **_kwargs: _evaluation( - contributions={"business_quality": "21.25"} + dashboard, "get_current_evaluation", + lambda *_args, **_kwargs: SimpleNamespace( + evaluation=_evaluation(contributions={"business_quality": "21.25"}), + fresh=False, last_verified_at=_SCORED_AT, + snapshot=SimpleNamespace(inputs=SimpleNamespace( + issue=issue, source_documents=(document.document_url,), + profile=SimpleNamespace(submitted_at=_SCORED_AT), + )), ), ) snapshot = build_dashboard_snapshot( now=newer, - session_factory=object, + session_factory=cast(SessionFactory, object), ) assert snapshot.rows[0].evaluation_stale is True diff --git a/tests/test_ipo_extraction_review.py b/tests/test_ipo_extraction_review.py index fe3efad..91c8ed9 100644 --- a/tests/test_ipo_extraction_review.py +++ b/tests/test_ipo_extraction_review.py @@ -1053,6 +1053,11 @@ def test_lost_approval_race_rolls_back_the_manual_revision( lambda *_args, **_kwargs: None, ) + from backend.storage.ipo_repository import get_ipo_scoring_state_values + + with file_session_factory() as session: + state_before = get_ipo_scoring_state_values(session, issue.id) + with pytest.raises(IpoValidationError, match="reviewed concurrently"): approve_extraction_proposal( proposal.id, @@ -1065,6 +1070,9 @@ def test_lost_approval_race_rolls_back_the_manual_revision( issue.id, session_factory=file_session_factory ) is None + with file_session_factory() as session: + assert get_ipo_scoring_state_values(session, issue.id) == state_before + def test_pending_proposal_blocks_document_deletion_but_reviewed_history_survives( file_session_factory, tmp_path: Path diff --git a/tests/test_ipo_models.py b/tests/test_ipo_models.py index 1d8fb0b..80e5cf5 100644 --- a/tests/test_ipo_models.py +++ b/tests/test_ipo_models.py @@ -232,6 +232,9 @@ def test_public_ipo_package_exports_the_domain_and_repository_contract() -> None "evaluate_issue", "fetch_sebi_filings", "get_document", + # Current selection adds freshness semantics without replacing the + # immutable history lookup that existing callers already import. + "get_current_evaluation", "get_evaluation", "get_financial", "get_issue", diff --git a/tests/test_ipo_scoring_service.py b/tests/test_ipo_scoring_service.py index a1b9145..3162da1 100644 --- a/tests/test_ipo_scoring_service.py +++ b/tests/test_ipo_scoring_service.py @@ -17,6 +17,8 @@ from pathlib import Path from typing import Any +import pytest + from backend.ipo.manual_extraction import ( IpoAmountUnit, IpoManualExtractionData, @@ -54,6 +56,19 @@ _AS_OF = dt.datetime(2026, 7, 13, 12, 0, tzinfo=dt.UTC) +def _input_revision(session, issue_id: int) -> int: + """Require the state fixture and return its authoritative scalar revision. + + Beginner note: + SQL scalar reads avoid identity-map caching when checking mutation tokens. + """ + from backend.storage.ipo_repository import get_ipo_scoring_state_values + + state = get_ipo_scoring_state_values(session, issue_id) + assert state is not None + return state[0] + + def _issue_data(**overrides: Any) -> IpoIssueData: """Build the reusable issue payload used by the scenarios below.""" values: dict[str, Any] = { @@ -225,6 +240,489 @@ def test_rescore_is_idempotent_until_an_input_changes( assert third.evaluation.inputs_fingerprint != first.evaluation.inputs_fingerprint +def test_return_to_a_selects_old_receipt_as_current(file_session_factory, tmp_path: Path) -> None: + """Beginner note: newest history incorrectly remained B after A was reused.""" + from backend.ipo.repository import get_latest_evaluation, list_evaluations + from backend.ipo.scoring import service + + issue = _scored_issue(file_session_factory, tmp_path) + first = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + update_issue(issue.id, _issue_data(price_band_high=Decimal("300")), session_factory=file_session_factory) + second = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + update_issue(issue.id, _issue_data(), session_factory=file_session_factory) + third = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert first.evaluation == third.evaluation + assert len(list_evaluations(issue.id, session_factory=file_session_factory)) == 2 + assert get_latest_evaluation(issue.id, session_factory=file_session_factory) == second.evaluation + assert hasattr(service, "get_current_evaluation"), "Need verified current selection separate from history" + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert current.evaluation == first.evaluation + assert current.fresh + assert current.last_verified_at is not None + + +def test_legacy_claimed_fingerprint_cannot_certify_invented_scorecard( + file_session_factory, tmp_path: Path, caplog +) -> None: + """Reject a fabricated legacy scorecard even when its fingerprint is genuine. + + Beginner note: + The old reuse path selected a caller's 100-point score instead of the + result derived from stored evidence. A failed verification must preserve + both the immutable history and the unverified current-selection state. + """ + from backend.ipo import repository + from backend.ipo.models import FactorAssessment, IpoValidationError + from backend.ipo.scoring import service + from backend.ipo.scoring.factor_derivation import derive_score_input + from backend.ipo.scoring.score_model import score_ipo + from backend.storage.ipo_repository import get_ipo_scoring_state_values + + issue = _scored_issue(file_session_factory, tmp_path) + snapshot = repository.load_ipo_scoring_snapshot(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + legitimate = derive_score_input(snapshot.inputs) + invented_factor = FactorAssessment(Decimal("100"), "Caller supplied assessment") + invented = dataclasses.replace( + legitimate, + business_quality=invented_factor, + financial_growth=invented_factor, + return_ratios=invented_factor, + valuation=invented_factor, + qib_subscription=invented_factor, + promoter_quality=invented_factor, + gmp_sentiment=invented_factor, + ) + legacy = repository.evaluate_issue(issue.id, invented, + inputs_fingerprint=compute_inputs_fingerprint(snapshot.inputs), model_version=SCREENER_MODEL_VERSION, + session_factory=file_session_factory) + assert legacy.result.score != score_ipo(legitimate).score + with file_session_factory() as session: + state_before = get_ipo_scoring_state_values(session, issue.id) + with pytest.raises(IpoValidationError, match="payload"): + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + with file_session_factory() as session: + assert get_ipo_scoring_state_values(session, issue.id) == state_before + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == [legacy] + assert not service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + assert not any("ipo_issue_scored" in record.message for record in caplog.records) + + +@pytest.mark.parametrize(("half", "field", "replacement"), [ + *(("score", name, Decimal("37.25")) for name in ( + "business_quality", "financial_growth", "return_ratios", "valuation", + "qib_subscription", "promoter_quality", "gmp_sentiment", "total_score", + )), + ("score", "contributions_json", {}), + ("score", "breakdown_json", []), + ("score", "missing_data_json", ["business_quality"]), + ("score", "reasons_json", ["Caller supplied score explanation"]), + ("recommendation", "recommendation", "Not Recommended"), + ("recommendation", "recommendation_type", "Skip"), + ("recommendation", "confidence", "high"), + ("recommendation", "reasons_json", ["Caller supplied verdict explanation"]), + ("recommendation", "missing_data_json", ["business_quality"]), + ("recommendation", "source_documents_json", []), + ("recommendation", "caution_flags_json", []), +]) +def test_reuse_compares_complete_receipt_not_only_total_or_identity( + file_session_factory, tmp_path: Path, half, field, replacement +) -> None: + """Reject each changed receipt component without certifying or rewriting it. + + Beginner note: + The old path compared identity alone. The fixture changes one field at + a time while retaining the fingerprint, proving that factors, verdicts, + reasons and source receipts all participate in verification. The chosen + replacement must actually differ from the fixture's stored value. + """ + from backend.ipo import repository + from backend.ipo.models import IpoValidationError + from backend.ipo.scoring.caution_flags import evaluate_caution_flags + from backend.ipo.scoring.factor_derivation import derive_score_input + from backend.storage.ipo_repository import get_ipo_evaluation_rows, get_ipo_scoring_state_values + + issue = _scored_issue(file_session_factory, tmp_path) + inputs = repository.load_ipo_factor_inputs_snapshot(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + legacy = repository.evaluate_issue(issue.id, derive_score_input(inputs), + caution_flags=evaluate_caution_flags(inputs), inputs_fingerprint=compute_inputs_fingerprint(inputs), + model_version=SCREENER_MODEL_VERSION, session_factory=file_session_factory) + with file_session_factory() as session: + rows = get_ipo_evaluation_rows(session, issue.id, legacy.score_id) + assert rows is not None + row = rows[0] if half == "score" else rows[1] + assert getattr(row, field) != replacement, "Fixture must change actual persisted semantics" + setattr(row, field, replacement) + history_before = repository.list_evaluations(issue.id, session_factory=file_session_factory) + with file_session_factory() as session: + state_before = get_ipo_scoring_state_values(session, issue.id) + with pytest.raises(IpoValidationError, match="payload"): + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == history_before + with file_session_factory() as session: + assert get_ipo_scoring_state_values(session, issue.id) == state_before + + +def test_matching_legacy_receipt_can_be_verified_with_reordered_source_set( + file_session_factory, tmp_path: Path +) -> None: + """Verify a matching historical pair despite harmless source-set reordering. + + Beginner note: + Refusing every old pair would break idempotency and A-B-A reuse. Source + URLs describe a set, so changing their order must still select the same + immutable receipt and leave its original calculation timestamp intact. + """ + from backend.ipo import repository + from backend.ipo.scoring import service + from backend.ipo.scoring.caution_flags import evaluate_caution_flags + from backend.ipo.scoring.factor_derivation import derive_score_input + + issue = _scored_issue(file_session_factory, tmp_path) + create_document(issue.id, IpoDocumentData(document_type="drhp", + document_url="https://www.sebi.gov.in/filings/older-drhp.html", source_confidence=Confidence.HIGH), + session_factory=file_session_factory) + inputs = repository.load_ipo_factor_inputs_snapshot(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + score_input = derive_score_input(inputs) + legacy = repository.evaluate_issue(issue.id, + dataclasses.replace(score_input, source_documents=tuple(reversed(score_input.source_documents))), + caution_flags=evaluate_caution_flags(inputs), inputs_fingerprint=compute_inputs_fingerprint(inputs), + model_version=SCREENER_MODEL_VERSION, session_factory=file_session_factory) + outcome = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert outcome.status == "skipped_unchanged" and outcome.evaluation == legacy + assert service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + + +def test_registered_sources_enter_semantic_identity(file_session_factory, tmp_path: Path) -> None: + """Beginner note: missing source URLs in the hash reused an incomplete receipt.""" + issue = _scored_issue(file_session_factory, tmp_path) + first = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + create_document(issue.id, IpoDocumentData(document_type="drhp", + document_url="https://www.sebi.gov.in/filings/older-drhp.html", source_confidence=Confidence.HIGH), + session_factory=file_session_factory) + second = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert second.status == "evaluated" + assert second.evaluation != first.evaluation + + +def test_dashboard_refresh_reuses_calculation_but_is_current(file_session_factory, tmp_path: Path) -> None: + """Beginner note: evidence newer than scored_at is valid after reverification.""" + from backend.ipo.dashboard import build_dashboard_snapshot + from backend.ipo.scoring.service import get_current_evaluation + + issue = _scored_issue(file_session_factory, tmp_path) + first = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + before = get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + update_issue(issue.id, _issue_data(), session_factory=file_session_factory) + assert not get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + second = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + after = get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert before.last_verified_at is not None and after.last_verified_at is not None + assert before.last_verified_at <= after.last_verified_at + assert first.evaluation is not None + assert first.evaluation == second.evaluation + row = build_dashboard_snapshot(now=_AS_OF, session_factory=file_session_factory).rows[0] + assert not row.evaluation_stale + assert row.calculated_at == first.evaluation.scored_at + assert row.last_verified_at == after.last_verified_at + + +def test_publication_reloads_changed_inputs_and_rolls_back_late_failure( + file_session_factory, tmp_path: Path, monkeypatch +) -> None: + """Beginner note: a CAS prevents committing a receipt for overwritten inputs.""" + from backend.ipo import repository + from backend.ipo.scoring import service + + issue = _scored_issue(file_session_factory, tmp_path) + real = service.derive_score_input + calls = [] + + def change_once(inputs): + """Commit newer evidence after the first snapshot, before publication.""" + calls.append(inputs.issue.price_band_high) + if len(calls) == 1: + update_issue(issue.id, _issue_data(price_band_high=Decimal("300")), session_factory=file_session_factory) + return real(inputs) + + monkeypatch.setattr(service, "derive_score_input", change_once) + outcome = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert calls == [Decimal("242"), Decimal("300")] + assert len(repository.list_evaluations(issue.id, session_factory=file_session_factory)) == 1 + assert service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + + update_issue(issue.id, _issue_data(price_band_high=Decimal("320")), session_factory=file_session_factory) + + def fail_selection(*args, **kwargs): + """Fail after pair insertion to prove the outer rollback includes history.""" + raise ValueError("late publication failure") + + monkeypatch.setattr(repository, "select_ipo_current_evaluation", fail_selection) + with pytest.raises(ValueError, match="late publication"): + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == [outcome.evaluation] + assert not service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + + +def test_snapshot_aba_mutation_and_total_retry_budget(file_session_factory, tmp_path: Path, monkeypatch) -> None: + """Beginner note: equal final prices cannot conceal an intervening A-B-A write.""" + from backend.ipo import repository + from backend.ipo.scoring.state import IpoScoringConflictError + + issue = _scored_issue(file_session_factory, tmp_path) + real = repository.get_latest_ipo_manual_extraction + calls = [] + + def change_during_read(session, issue_id): + """Restore the same price through two commits while input reads are open.""" + profile = real(session, issue_id) + calls.append(issue_id) + update_issue(issue.id, _issue_data(price_band_high=Decimal("300")), session_factory=file_session_factory) + update_issue(issue.id, _issue_data(), session_factory=file_session_factory) + return profile + + monkeypatch.setattr(repository, "get_latest_ipo_manual_extraction", change_during_read) + with pytest.raises(IpoScoringConflictError, match="three attempts"): + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert calls == [issue.id] * 3 + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == [] + + +def test_same_input_concurrent_publication_is_idempotent(file_session_factory, tmp_path: Path, monkeypatch) -> None: + """Beginner note: real concurrent transactions must retain exactly one pair.""" + from concurrent.futures import ThreadPoolExecutor + from threading import Barrier + + from backend.ipo import repository + from backend.ipo.scoring import service + + issue = _scored_issue(file_session_factory, tmp_path) + barrier = Barrier(2) + real = service.load_ipo_scoring_snapshot + + def synchronized_snapshot(*args, **kwargs): + """Release both real scoring transactions with the same detached snapshot.""" + snapshot = real(*args, **kwargs) + barrier.wait(timeout=15) + return snapshot + + monkeypatch.setattr(service, "load_ipo_scoring_snapshot", synchronized_snapshot) + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [pool.submit(rescore_issue, issue.id, as_of=_AS_OF, session_factory=file_session_factory) + for _ in range(2)] + outcomes = [future.result(timeout=30) for future in futures] + assert {outcome.status for outcome in outcomes} == {"evaluated", "skipped_unchanged"} + assert outcomes[0].evaluation == outcomes[1].evaluation + assert len(repository.list_evaluations(issue.id, session_factory=file_session_factory)) == 1 + + +def test_snapshot_and_publication_share_three_attempts(file_session_factory, tmp_path: Path, monkeypatch) -> None: + """Beginner note: separate retry loops could multiply the promised three-attempt limit.""" + from backend.ipo import repository + from backend.ipo.scoring import service + from backend.ipo.scoring.state import IpoScoringConflictError + + issue = _scored_issue(file_session_factory, tmp_path) + previous = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + real_snapshot = service.load_ipo_scoring_snapshot + real_derive = service.derive_score_input + snapshots, publications = [], [] + + def conflicting_snapshot(*args, **kwargs): + """Spend the first attempt on assembly, leaving only two publication tries.""" + snapshots.append(1) + if len(snapshots) == 1: + raise IpoScoringConflictError("snapshot conflict") + return real_snapshot(*args, **kwargs) + + def conflicting_calculation(inputs): + """Commit a real input mutation before each attempted publication.""" + publications.append(1) + update_issue(issue.id, _issue_data(price_band_high=Decimal(300 + len(publications))), + session_factory=file_session_factory) + return real_derive(inputs) + + monkeypatch.setattr(service, "load_ipo_scoring_snapshot", conflicting_snapshot) + monkeypatch.setattr(service, "derive_score_input", conflicting_calculation) + with pytest.raises(IpoScoringConflictError, match="three attempts"): + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert len(snapshots) == 3 and len(publications) == 2 + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == [previous.evaluation] + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert not current.fresh and current.evaluation == previous.evaluation + + +def test_clock_only_expiry_near_close_and_model_freshness(file_session_factory, tmp_path: Path, monkeypatch) -> None: + """Beginner note: clock/model changes invalidate receipts without DB mutations.""" + from backend.ipo.scoring import service + + issue = _scored_issue(file_session_factory, tmp_path) + signal = IpoEnrichmentSignalData(signal_type=IpoEnrichmentSignalType.GMP, captured_at=_AS_OF, + query_text="Example Ltd IPO GMP", payload=({"title": "GMP report"},), parsed_value=Decimal("25"), + quarantined=False, confidence=Confidence.LOW, source_policy="serpapi-low-confidence-v2") + record_enrichment_signals(issue.id, [signal], session_factory=file_session_factory) + first = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + threshold = _AS_OF + dt.timedelta(days=5) + assert service.get_current_evaluation(issue.id, as_of=threshold, session_factory=file_session_factory).fresh + assert not service.get_current_evaluation(issue.id, as_of=threshold + dt.timedelta(microseconds=1), + session_factory=file_session_factory).fresh + rescore_issue(issue.id, as_of=threshold + dt.timedelta(microseconds=1), session_factory=file_session_factory) + record_enrichment_signals(issue.id, [dataclasses.replace(signal, captured_at=threshold)], + session_factory=file_session_factory) + restored = rescore_issue(issue.id, as_of=threshold, session_factory=file_session_factory) + assert restored.evaluation == first.evaluation + monkeypatch.setattr(service, "SCREENER_MODEL_VERSION", "future-model") + current = service.get_current_evaluation(issue.id, as_of=threshold, session_factory=file_session_factory) + assert current.reason == "model_changed" + + +def test_production_clock_crossing_rolls_back_first_pair(file_session_factory, tmp_path: Path, monkeypatch) -> None: + """Beginner note: input CAS alone misses crossing midnight during calculation.""" + from backend.ipo import repository + from backend.ipo.scoring import service + + issue = _scored_issue(file_session_factory, tmp_path) + update_issue(issue.id, _issue_data(status=IpoStatus.OPEN, close_date=dt.date(2026, 7, 15)), + session_factory=file_session_factory) + before = dt.datetime(2026, 7, 13, 23, 59, 59, tzinfo=dt.UTC) + after = before + dt.timedelta(seconds=1) + clocks = iter([before, after, after, after]) + monkeypatch.setattr(service, "_utc_now", lambda: next(clocks)) + outcome = rescore_issue(issue.id, session_factory=file_session_factory) + assert repository.list_evaluations(issue.id, session_factory=file_session_factory) == [outcome.evaluation] + assert service.get_current_evaluation(issue.id, as_of=after, session_factory=file_session_factory).fresh + assert not service.get_current_evaluation(issue.id, as_of=before, session_factory=file_session_factory).fresh + with pytest.raises(ValueError, match="timezone-aware"): + rescore_issue(issue.id, as_of=before.replace(tzinfo=None), session_factory=file_session_factory) + + +@pytest.mark.parametrize("operation", [ + "issue", "document_insert", "document_values", "document_row", "document_delete", + "subscription_insert", "subscription_update", "subscription_delete", +]) +def test_storage_writers_invalidate_and_roll_back_atomically(file_session_factory, tmp_path: Path, operation) -> None: + """Beginner note: facade-only invalidation missed ingestion and rollback paths.""" + from backend.storage import ipo_repository as storage + + issue = _scored_issue(file_session_factory, tmp_path) + with file_session_factory() as session: + document = storage.insert_ipo_document(session, issue.id, { + "document_type": "drhp", "document_url": "https://www.sebi.gov.in/delete-me", + "source_confidence": "high", + }) + document_id = document.id + subscription_id = storage.insert_ipo_subscription(session, issue.id, { + "captured_at": _AS_OF, "qib_multiple": Decimal("20"), "source_confidence": "high", + }).id + before = _input_revision(session, issue.id) + + def mutate(session): + """Exercise production SQL writers, including ingestion's direct helper.""" + if operation == "issue": + storage.update_ipo_issue_row(session, issue.id, {"price_band_high": Decimal("300")}) + elif operation == "document_insert": + storage.insert_ipo_document(session, issue.id, {"document_type": "drhp", + "document_url": "https://www.sebi.gov.in/new-source", "source_confidence": "high"}) + elif operation == "document_values": + existing_document = storage.get_ipo_document(session, issue.id, document_id) + assert existing_document is not None + storage.update_ipo_document_values( + session, existing_document, {"document_url": "https://www.sebi.gov.in/revised-source"} + ) + elif operation == "document_row": + storage.update_ipo_document_row(session, issue.id, document_id, {"document_type": "rhp"}) + elif operation == "document_delete": + storage.delete_ipo_document_row(session, issue.id, document_id) + elif operation == "subscription_insert": + storage.insert_ipo_subscription(session, issue.id, {"captured_at": _AS_OF + dt.timedelta(hours=1), + "qib_multiple": Decimal("30"), "source_confidence": "high"}) + elif operation == "subscription_update": + storage.update_ipo_subscription_row(session, issue.id, subscription_id, {"qib_multiple": Decimal("30")}) + else: + storage.delete_ipo_subscription_row(session, issue.id, subscription_id) + + with pytest.raises(RuntimeError, match="abort"), file_session_factory() as session: + mutate(session) + assert _input_revision(session, issue.id) > before + raise RuntimeError("abort") + with file_session_factory() as session: + assert _input_revision(session, issue.id) == before + stored_issue = storage.get_ipo_issue(session, issue.id) + stored_document = storage.get_ipo_document(session, issue.id, document_id) + stored_subscription = storage.get_ipo_subscription(session, issue.id, subscription_id) + assert stored_issue is not None and stored_issue.price_band_high == Decimal("242") + assert stored_document is not None and stored_document.document_type == "drhp" + assert stored_subscription is not None and stored_subscription.qib_multiple == Decimal("20") + mutate(session) + with file_session_factory() as session: + assert _input_revision(session, issue.id) > before + + +def test_manual_and_enrichment_writers_invalidate_but_cache_does_not(file_session_factory, tmp_path: Path) -> None: + """Beginner note: every approved profile and real last-seen refresh needs verification.""" + from backend.ipo import repository + from backend.ipo.scoring import service + from backend.storage import ipo_repository as storage + + issue = _scored_issue(file_session_factory, tmp_path) + document = repository.list_documents(issue.id, session_factory=file_session_factory)[0] + first = service.rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + before = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + with file_session_factory() as session: + stored_document = storage.get_ipo_document(session, issue.id, document.id) + assert stored_document is not None + storage.update_ipo_document_values(session, stored_document, + {"downloaded_at": _AS_OF + dt.timedelta(days=100)}) + assert service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + submit_manual_extraction(issue.id, _profile_data(document.id), entered_by_email="admin@example.com", + data_dir=tmp_path, session_factory=file_session_factory) + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert not current.fresh + assert current.snapshot.state.input_revision > before.snapshot.state.input_revision + assert rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory).evaluation == first.evaluation + signal = IpoEnrichmentSignalData(signal_type=IpoEnrichmentSignalType.GMP, captured_at=_AS_OF, + query_text="Example Ltd IPO GMP", payload=({"title": "GMP report"},), parsed_value=Decimal("25"), + quarantined=False, confidence=Confidence.LOW, source_policy="serpapi-low-confidence-v2") + record_enrichment_signals(issue.id, [signal], session_factory=file_session_factory) + rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + record_enrichment_signals(issue.id, [dataclasses.replace(signal, captured_at=_AS_OF + dt.timedelta(hours=1))], + session_factory=file_session_factory) + assert not service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory).fresh + with file_session_factory() as session: + existing = storage.list_ipo_enrichment_signal_rows(session, issue.id)[0] + values = {column.name: getattr(existing, column.name) for column in existing.__table__.columns + if column.name not in {"id", "issue_id"}} + values["semantic_hash"] = "a" * 64 + values["captured_at"] = _AS_OF + dt.timedelta(hours=2) + before_revision = _input_revision(session, issue.id) + storage.insert_ipo_enrichment_signals(session, issue.id, [values]) + assert _input_revision(session, issue.id) > before_revision + + +def test_current_pointer_ownership_and_score_deletion_fail_closed(file_session_factory, tmp_path: Path) -> None: + """Beginner note: the score FK alone cannot prove that a pointer belongs to its issue.""" + from backend.ipo import repository + from backend.ipo.scoring import service + from backend.storage.models import IpoScoringState + + issue = _scored_issue(file_session_factory, tmp_path) + other = create_issue(_issue_data(company_name="Other Ltd"), session_factory=file_session_factory) + result = rescore_issue(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert result.evaluation is not None + with file_session_factory() as session: + state = session.get(IpoScoringState, other.id) + state.current_score_id = result.evaluation.score_id + state.evaluated_revision = state.input_revision + state.last_verified_at = _AS_OF + current = service.get_current_evaluation(other.id, as_of=_AS_OF, session_factory=file_session_factory) + assert not current.fresh and current.evaluation is None + assert repository.delete_evaluation(issue.id, result.evaluation.score_id, session_factory=file_session_factory) + current = service.get_current_evaluation(issue.id, as_of=_AS_OF, session_factory=file_session_factory) + assert not current.fresh and current.snapshot.state.current_score_id is None + repository.delete_issue(other.id, session_factory=file_session_factory) + with file_session_factory() as session: + assert session.get(IpoScoringState, other.id) is None + + def test_new_subscription_and_enrichment_change_the_fingerprint( file_session_factory, tmp_path: Path ) -> None: diff --git a/tests/test_ipo_screener_module.py b/tests/test_ipo_screener_module.py index 785428f..aa0dd9a 100644 --- a/tests/test_ipo_screener_module.py +++ b/tests/test_ipo_screener_module.py @@ -52,6 +52,16 @@ def _row(**overrides: Any) -> IpoDashboardRow: return IpoDashboardRow(**values) +def test_stale_history_is_not_an_actionable_rating() -> None: + """Beginner note: the adapter previously exported stale history as a live rating.""" + row = _row(evaluation_stale=True) + result = ipo_screener._result_row(row, scanner=ipo_screener.IpoScreener(), failed=False) + assert result["rating"] is None + assert result["recommendation_type"] is None + assert result["signal_date"] is None + assert result["historical_recommendation"] == row.recommendation + + def _install(monkeypatch, rows: list[IpoDashboardRow], **overrides: Any) -> dict[str, Any]: """Fake the pipeline, auto-approval, and snapshot; capture the call args. diff --git a/tests/test_run_ipo_screener_job.py b/tests/test_run_ipo_screener_job.py index 8c14db6..da11e3a 100644 --- a/tests/test_run_ipo_screener_job.py +++ b/tests/test_run_ipo_screener_job.py @@ -389,6 +389,32 @@ def _downloader(*_args: Any, **_kwargs: Any) -> None: assert result.exit_code == 1 +def test_scoring_conflict_is_retryable_failure_and_keeps_sibling_success() -> None: + """Beginner note: exhausted publication conflicts cannot count as a current score.""" + from backend.ipo.scoring.state import IpoScoringConflictError + + issues = [_issue(1, "Acme Ltd"), _issue(2, "Beta Ltd")] + + def rescorer(issue_id, **kwargs): + """Simulate bounded conflict exhaustion, followed by a successful sibling.""" + if issue_id == 1: + raise IpoScoringConflictError("untrusted details must not be printed") + return _rescore(issues[1], "skipped_unchanged", _evaluation( + score="70.00", recommendation="Recommended", recommendation_type=APPLY_AND_HOLD, + )) + + out = io.StringIO() + result = run_ipo_screener(skip_scan=True, skip_download=True, skip_enrich=True, + ensure_schema=lambda: True, issue_lister=lambda **kwargs: issues, + document_lister=lambda *args, **kwargs: [], rescorer=rescorer, session_factory=object, output=out) + assert result.issues[0].status == "failed" + assert result.issues[0].retryable is True + assert result.issues[0].failure_code == "ipo_scoring_conflict" + assert result.issues[1].status == "skipped_unchanged" + assert result.exit_code == 1 + assert "untrusted details" not in out.getvalue() + + def test_missing_serpapi_key_is_a_graceful_skip_not_a_failure() -> None: """The first no-key outcome stops further queries and stays exit 0.""" issues = [_issue(1, "Acme Ltd"), _issue(2, "Beta Ltd")] diff --git a/tests/test_scan_storage_migrations.py b/tests/test_scan_storage_migrations.py index 92290c6..08906f6 100644 --- a/tests/test_scan_storage_migrations.py +++ b/tests/test_scan_storage_migrations.py @@ -17,6 +17,7 @@ import os import subprocess import sys +from decimal import Decimal from pathlib import Path import pytest @@ -29,6 +30,62 @@ from backend.storage.models import Base, IpoIssue, IpoManualExtraction, IpoScore +def test_ipo013_backfills_complete_history_without_certifying_it(monkeypatch, tmp_path: Path) -> None: + """Beginner note: newest orphan/legacy receipts must never become verified state.""" + from sqlalchemy.exc import IntegrityError + + from backend.storage.models import IpoRecommendation + + database_url = f"sqlite:///{(tmp_path / 'ipo013.db').as_posix()}" + monkeypatch.setenv("DATABASE_URL", database_url) + config = Config("alembic.ini") + command.upgrade(config, "20260909valid005") + engine = database._make_engine(database_url) + with Session(engine) as session: + issue = IpoIssue(company_name="Legacy Ltd", issue_type="mainboard", status="rhp_filed", + source_confidence="high") + empty = IpoIssue(company_name="Empty Ltd", issue_type="mainboard", status="rhp_filed", + source_confidence="high") + session.add_all([issue, empty]) + session.flush() + issue_id, empty_id = issue.id, empty.id + tied = dt.datetime(2026, 9, 1, tzinfo=dt.UTC) + scores = [] + for ordinal in range(3): + score = IpoScore(issue_id=issue.id, total_score=Decimal("50"), contributions_json={}, + missing_data_json=[], reasons_json=["preserve original"], model_version="legacy", + scored_at=tied if ordinal < 2 else tied + dt.timedelta(days=1)) + session.add(score) + session.flush() + scores.append(score.id) + if ordinal < 2: + session.add(IpoRecommendation(score_id=score.id, recommendation="Not Recommended", + recommendation_type="Skip", confidence="low", reasons_json=["preserve recommendation"], + missing_data_json=[], source_documents_json=["https://www.sebi.gov.in/legacy"])) + session.commit() + with engine.connect() as connection: + original = list(connection.execute(text("SELECT * FROM ipo_scores ORDER BY id"))) + receipts = list(connection.execute(text("SELECT * FROM ipo_recommendations ORDER BY id"))) + command.upgrade(config, "head") + with engine.connect() as connection: + states = list(connection.execute(text("SELECT * FROM ipo_scoring_state ORDER BY issue_id"))) + assert states == [(issue_id, 0, scores[1], None, None), (empty_id, 0, None, None, None)] + assert list(connection.execute(text("SELECT * FROM ipo_scores ORDER BY id"))) == original + assert list(connection.execute(text("SELECT * FROM ipo_recommendations ORDER BY id"))) == receipts + for assignment in ("input_revision=-1", "evaluated_revision=-1", "evaluated_revision=1"): + with pytest.raises(IntegrityError), engine.begin() as connection: + connection.execute(text(f"UPDATE ipo_scoring_state SET {assignment}")) + with engine.begin() as connection: + connection.execute(text("DELETE FROM ipo_scores WHERE id=:id"), {"id": scores[1]}) + assert connection.execute(text("SELECT current_score_id FROM ipo_scoring_state WHERE issue_id=:id"), + {"id": issue_id}).scalar_one() is None + connection.execute(text("DELETE FROM ipo_issues WHERE id=:id"), {"id": issue_id}) + assert connection.execute(text("SELECT COUNT(*) FROM ipo_scoring_state")).scalar_one() == 1 + command.downgrade(config, "20260909valid005") + assert "ipo_scoring_state" not in inspect(engine).get_table_names() + engine.dispose() + + def test_obs004a_backfills_legacy_rows_and_restores_original_shape( monkeypatch, tmp_path: Path ): @@ -147,6 +204,7 @@ def test_alembic_upgrade_and_downgrade_use_temp_sqlite(monkeypatch, tmp_path: Pa "ipo_manual_peer_valuations", "ipo_recommendations", "ipo_scores", + "ipo_scoring_state", "ipo_subscriptions", "scan_runs", "scan_results", @@ -817,6 +875,7 @@ def test_ensure_database_schema_creates_tables_and_short_circuits(monkeypatch, t "ipo_manual_peer_valuations", "ipo_recommendations", "ipo_scores", + "ipo_scoring_state", "ipo_subscriptions", "scan_runs", "scan_results", diff --git a/ui/ipo_page.py b/ui/ipo_page.py index f4b673b..89adff4 100644 --- a/ui/ipo_page.py +++ b/ui/ipo_page.py @@ -60,14 +60,13 @@ _VERDICT_FILTERS = ("All", "Recommended", "Not Recommended") -@st.cache_data(ttl=300, show_spinner=False) def _load_snapshot() -> IpoDashboardSnapshot: - """Read (and briefly cache) the dashboard snapshot for this session. + """Read current evidence at every render without a global snapshot cache. Beginner note: - ``st.cache_data`` keeps Streamlit's rerun-per-interaction model cheap: - clicking a filter does not re-read every issue. The re-score handler - calls ``.clear()`` so its fresh evaluations appear immediately. + Other sessions and jobs can write while this page is open, and GMP + can expire without a write. A fixed TTL cannot certify actionable + freshness. Each rerun rebuilds with one shared UTC evaluation clock. """ return build_dashboard_snapshot() @@ -76,7 +75,8 @@ def _verdict_label(row: IpoDashboardRow) -> str: """Map one stored recommendation_type onto its display wording.""" if row.recommendation_type is None: return "Not scored yet" - return _RECOMMENDATION_TYPE_LABELS.get(row.recommendation_type, row.recommendation_type) + label = _RECOMMENDATION_TYPE_LABELS.get(row.recommendation_type, row.recommendation_type) + return f"Historical: {label}" if row.evaluation_stale else label def _apply_verdict_filter( @@ -85,7 +85,7 @@ def _apply_verdict_filter( """Narrow rows to one binary verdict; unscored rows only pass 'All'.""" if choice == "All": return rows - return tuple(row for row in rows if row.recommendation == choice) + return tuple(row for row in rows if not row.evaluation_stale and row.recommendation == choice) def _rows_frame(rows: tuple[IpoDashboardRow, ...]) -> pd.DataFrame: @@ -127,6 +127,8 @@ def _rows_frame(rows: tuple[IpoDashboardRow, ...]) -> pd.DataFrame: "Source documents": "; ".join(row.source_documents), "Last updated": row.last_updated.isoformat() if row.last_updated else "", "Evaluation stale": row.evaluation_stale, + "Calculated at": row.calculated_at.isoformat() if row.calculated_at else "", + "Last verified at": row.last_verified_at.isoformat() if row.last_verified_at else "", } for row in rows ] @@ -205,8 +207,12 @@ def _render_breakdowns(rows: tuple[IpoDashboardRow, ...]) -> None: ) if row.evaluation_stale: st.caption( - "Evaluation stale: newer evidence is waiting to be scored." + "Historical evaluation: re-score to check current evidence and eligibility." ) + if row.calculated_at: + st.caption(f"Calculated at {row.calculated_at.isoformat()}.") + if row.last_verified_at: + st.caption(f"Last verified at {row.last_verified_at.isoformat()}.") if row.source_documents: st.caption( "Source documents: " @@ -254,7 +260,7 @@ def _render_ipo_page(*, can_rescore: bool, user_email: str | None = None) -> Non Beginner note: Rendering is read-only. The one explicit button calls the shared - repository-only scoring service, invalidates the short snapshot cache, + repository-only scoring service, rebuilds the current snapshot, and records an attributable audit event. """ st.subheader("IPO dashboard") @@ -266,7 +272,6 @@ def _render_ipo_page(*, can_rescore: bool, user_email: str | None = None) -> Non if can_rescore and st.button("Re-score all issues", key="ipo_rescore_all"): counts = _run_rescore_all(_load_snapshot(), user_email) - _load_snapshot.clear() st.success( "Re-score complete: " f"{counts['evaluated']} evaluated, "