busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
busya
2026-04-24 17:10:02 +03:00
parent b7a9ef71b2
commit 0b227a307c
34 changed files with 5220 additions and 5446 deletions
Download Patch File Download Diff File Expand all files Collapse all files

470
backend/src/services/dataset_review/clarification_engine.py
View File

@@ -3,20 +3,22 @@
# @SEMANTICS: dataset_review, clarification, question_payload, answer_persistence, readiness, findings
# @PURPOSE: Manage one-question-at-a-time clarification state, deterministic answer persistence, and readiness/finding updates.
# @LAYER: Domain
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [DEPENDS_ON] ->[ClarificationSession]
# @RELATION: [DEPENDS_ON] ->[ClarificationQuestion]
# @RELATION: [DEPENDS_ON] ->[ClarificationAnswer]
# @RELATION: [DEPENDS_ON] ->[ValidationFinding]
# @RELATION: DEPENDS_ON -> [DatasetReviewSessionRepository]
# @RELATION: DEPENDS_ON -> [ClarificationSession]
# @RELATION: DEPENDS_ON -> [ClarificationQuestion]
# @RELATION: DEPENDS_ON -> [ClarificationAnswer]
# @RELATION: DEPENDS_ON -> [ValidationFinding]
# @RELATION: DISPATCHES -> [ClarificationHelpers:Module]
# @PRE: Target session contains a persisted clarification aggregate in the current ownership scope.
# @POST: Active clarification payload exposes one highest-priority unresolved question, and each recorded answer is persisted before pointer/readiness mutation.
# @SIDE_EFFECT: Persists clarification answers, question/session states, and related readiness/finding changes.
# @DATA_CONTRACT: Input[DatasetReviewSession|ClarificationAnswerCommand] -> Output[ClarificationStateResult]
# @INVARIANT: Only one active clarification question may exist at a time; skipped and expert-review items remain unresolved and visible.
# @RATIONALE: Original 635-line file exceeded INV_7 (400-line module limit). Extracted pure helpers into _helpers sub-module.
# @REJECTED: Keeping all clarification logic in one file because it exceeded the fractal limit.
from __future__ import annotations
# [DEF:imports:Block]
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
@@ -30,19 +32,25 @@ from src.models.dataset_review import (
ClarificationSession,
ClarificationStatus,
DatasetReviewSession,
FindingArea,
FindingSeverity,
QuestionState,
ReadinessState,
RecommendedAction,
ResolutionState,
SessionPhase,
ValidationFinding,
)
from src.services.dataset_review.repositories.session_repository import (
DatasetReviewSessionRepository,
)
# [/DEF:imports:Block]
from src.services.dataset_review.clarification_pkg._helpers import (
select_next_open_question,
count_resolved_questions,
count_remaining_questions,
normalize_answer_value,
build_impact_summary,
upsert_clarification_finding,
derive_readiness_state,
derive_recommended_action,
)
# [DEF:ClarificationQuestionPayload:Class]
@@ -96,9 +104,8 @@ class ClarificationAnswerCommand:
# [DEF:ClarificationEngine:Class]
# @COMPLEXITY: 4
# @PURPOSE: Provide deterministic one-question-at-a-time clarification selection and answer persistence.
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
# @RELATION: [DEPENDS_ON] ->[ClarificationSession]
# @RELATION: [DEPENDS_ON] ->[ValidationFinding]
# @RELATION: DEPENDS_ON -> [DatasetReviewSessionRepository]
# @RELATION: CALLS -> [ClarificationHelpers:Module]
# @PRE: Repository is bound to the current request transaction scope.
# @POST: Returned clarification state is persistence-backed and aligned with session readiness/recommended action.
# @SIDE_EFFECT: Mutates clarification answers, session flags, and related clarification findings.
@@ -113,51 +120,33 @@ class ClarificationEngine:
# [DEF:build_question_payload:Function]
# @COMPLEXITY: 4
# @PURPOSE: Return the one active highest-priority clarification question payload with why-it-matters, current guess, and options.
# @RELATION: [DEPENDS_ON] ->[ClarificationQuestion]
# @RELATION: [DEPENDS_ON] ->[ClarificationOption]
# @PURPOSE: Return the one active highest-priority clarification question payload.
# @PRE: Session contains unresolved clarification state or a resumable clarification session.
# @POST: Returns exactly one active/open question payload or None when no unresolved question remains.
# @SIDE_EFFECT: Normalizes the active-question pointer and clarification status in persistence.
# @DATA_CONTRACT: Input[DatasetReviewSession] -> Output[ClarificationQuestionPayload|None]
def build_question_payload(
self,
session: DatasetReviewSession,
self, session: DatasetReviewSession,
) -> Optional[ClarificationQuestionPayload]:
with belief_scope("ClarificationEngine.build_question_payload"):
clarification_session = self._get_latest_clarification_session(session)
if clarification_session is None:
logger.reason(
"Clarification payload requested without clarification session",
extra={"session_id": session.session_id},
)
logger.reason("No clarification session found", extra={"session_id": session.session_id})
return None
active_questions = [
question
for question in clarification_session.questions
if question.state == QuestionState.OPEN
q for q in clarification_session.questions if q.state == QuestionState.OPEN
]
active_questions.sort(
key=lambda item: (
-int(item.priority),
item.created_at,
item.question_id,
)
)
active_questions.sort(key=lambda item: (-int(item.priority), item.created_at, item.question_id))
if not active_questions:
clarification_session.current_question_id = None
clarification_session.status = ClarificationStatus.COMPLETED
session.readiness_state = self._derive_readiness_state(session)
session.recommended_action = self._derive_recommended_action(session)
session.readiness_state = derive_readiness_state(session, clarification_session)
session.recommended_action = derive_recommended_action(session, clarification_session)
if session.current_phase == SessionPhase.CLARIFICATION:
session.current_phase = SessionPhase.REVIEW
self.repository.db.commit()
logger.reflect(
"No unresolved clarification question remains",
extra={"session_id": session.session_id},
)
logger.reflect("No unresolved clarification question remains", extra={"session_id": session.session_id})
return None
selected_question = active_questions[0]
@@ -167,15 +156,7 @@ class ClarificationEngine:
session.recommended_action = RecommendedAction.ANSWER_NEXT_QUESTION
session.current_phase = SessionPhase.CLARIFICATION
logger.reason(
"Selected active clarification question",
extra={
"session_id": session.session_id,
"clarification_session_id": clarification_session.clarification_session_id,
"question_id": selected_question.question_id,
"priority": selected_question.priority,
},
)
logger.reason("Selected active clarification question", extra={"session_id": session.session_id, "question_id": selected_question.question_id, "priority": selected_question.priority})
self.repository.db.commit()
payload = ClarificationQuestionPayload(
@@ -188,124 +169,58 @@ class ClarificationEngine:
priority=selected_question.priority,
state=selected_question.state,
options=[
{
"option_id": option.option_id,
"question_id": option.question_id,
"label": option.label,
"value": option.value,
"is_recommended": option.is_recommended,
"display_order": option.display_order,
}
for option in sorted(
selected_question.options,
key=lambda item: (
item.display_order,
item.label,
item.option_id,
),
)
{"option_id": o.option_id, "question_id": o.question_id, "label": o.label, "value": o.value, "is_recommended": o.is_recommended, "display_order": o.display_order}
for o in sorted(selected_question.options, key=lambda item: (item.display_order, item.label, item.option_id))
],
)
logger.reflect(
"Clarification payload built",
extra={
"session_id": session.session_id,
"question_id": payload.question_id,
"option_count": len(payload.options),
},
)
logger.reflect("Clarification payload built", extra={"session_id": session.session_id, "question_id": payload.question_id, "option_count": len(payload.options)})
return payload
# [/DEF:build_question_payload:Function]
# [DEF:record_answer:Function]
# @COMPLEXITY: 4
# @PURPOSE: Persist one clarification answer before any pointer/readiness mutation and compute deterministic state impact.
# @RELATION: [DEPENDS_ON] ->[ClarificationAnswer]
# @RELATION: [DEPENDS_ON] ->[ValidationFinding]
# @PURPOSE: Persist one clarification answer before any pointer/readiness mutation.
# @PRE: Target question belongs to the session's active clarification session and is still open.
# @POST: Answer row is persisted before current-question pointer advances; skipped/expert-review items remain unresolved and visible.
# @POST: Answer row is persisted before current-question pointer advances.
# @SIDE_EFFECT: Inserts answer row, mutates question/session states, updates clarification findings, and commits.
# @DATA_CONTRACT: Input[ClarificationAnswerCommand] -> Output[ClarificationStateResult]
def record_answer(
self, command: ClarificationAnswerCommand
) -> ClarificationStateResult:
def record_answer(self, command: ClarificationAnswerCommand) -> ClarificationStateResult:
with belief_scope("ClarificationEngine.record_answer"):
session = command.session
clarification_session = self._get_latest_clarification_session(session)
if clarification_session is None:
logger.explore(
"Cannot record clarification answer because no clarification session exists",
extra={"session_id": session.session_id},
)
logger.explore("Cannot record clarification answer because no clarification session exists", extra={"session_id": session.session_id})
raise ValueError("Clarification session not found")
question = self._find_question(clarification_session, command.question_id)
if question is None:
logger.explore(
"Cannot record clarification answer for foreign or missing question",
extra={
"session_id": session.session_id,
"question_id": command.question_id,
},
)
logger.explore("Cannot record clarification answer for foreign or missing question", extra={"session_id": session.session_id, "question_id": command.question_id})
raise ValueError("Clarification question not found")
if question.answer is not None:
logger.explore(
"Rejected duplicate clarification answer submission",
extra={
"session_id": session.session_id,
"question_id": command.question_id,
},
)
logger.explore("Rejected duplicate clarification answer submission", extra={"session_id": session.session_id, "question_id": command.question_id})
raise ValueError("Clarification question already answered")
if (
clarification_session.current_question_id
and clarification_session.current_question_id != question.question_id
):
logger.explore(
"Rejected answer for non-active clarification question",
extra={
"session_id": session.session_id,
"question_id": question.question_id,
"current_question_id": clarification_session.current_question_id,
},
)
raise ValueError(
"Only the active clarification question can be answered"
)
if clarification_session.current_question_id and clarification_session.current_question_id != question.question_id:
logger.explore("Rejected answer for non-active clarification question", extra={"session_id": session.session_id, "question_id": question.question_id, "current_question_id": clarification_session.current_question_id})
raise ValueError("Only the active clarification question can be answered")
normalized_answer_value = self._normalize_answer_value(
command.answer_kind, command.answer_value, question
)
normalized_answer_value = normalize_answer_value(command.answer_kind, command.answer_value, question)
logger.reason(
"Persisting clarification answer before state advancement",
extra={
"session_id": session.session_id,
"question_id": question.question_id,
"answer_kind": command.answer_kind.value,
},
)
logger.reason("Persisting clarification answer before state advancement", extra={"session_id": session.session_id, "question_id": question.question_id, "answer_kind": command.answer_kind.value})
persisted_answer = ClarificationAnswer(
question_id=question.question_id,
answer_kind=command.answer_kind,
answer_value=normalized_answer_value,
answered_by_user_id=command.user.id,
impact_summary=self._build_impact_summary(
question, command.answer_kind, normalized_answer_value
),
impact_summary=build_impact_summary(question, command.answer_kind, normalized_answer_value),
)
self.repository.db.add(persisted_answer)
self.repository.db.flush()
changed_finding = self._upsert_clarification_finding(
session=session,
question=question,
answer_kind=command.answer_kind,
answer_value=normalized_answer_value,
changed_finding = upsert_clarification_finding(
session=session, question=question, answer_kind=command.answer_kind,
answer_value=normalized_answer_value, db_session=self.repository.db,
)
if command.answer_kind == AnswerKind.SELECTED:
@@ -320,51 +235,26 @@ class ClarificationEngine:
question.updated_at = datetime.utcnow()
self.repository.db.flush()
clarification_session.resolved_count = self._count_resolved_questions(
clarification_session
)
clarification_session.remaining_count = self._count_remaining_questions(
clarification_session
)
clarification_session.summary_delta = self.summarize_progress(
clarification_session
)
clarification_session.resolved_count = count_resolved_questions(clarification_session)
clarification_session.remaining_count = count_remaining_questions(clarification_session)
clarification_session.summary_delta = self.summarize_progress(clarification_session)
clarification_session.updated_at = datetime.utcnow()
next_question = self._select_next_open_question(clarification_session)
clarification_session.current_question_id = (
next_question.question_id if next_question else None
)
clarification_session.status = (
ClarificationStatus.ACTIVE
if next_question
else ClarificationStatus.COMPLETED
)
next_question = select_next_open_question(clarification_session)
clarification_session.current_question_id = next_question.question_id if next_question else None
clarification_session.status = ClarificationStatus.ACTIVE if next_question else ClarificationStatus.COMPLETED
if clarification_session.status == ClarificationStatus.COMPLETED:
clarification_session.completed_at = datetime.utcnow()
session.readiness_state = self._derive_readiness_state(session)
session.recommended_action = self._derive_recommended_action(session)
session.current_phase = (
SessionPhase.CLARIFICATION
if clarification_session.current_question_id
else SessionPhase.REVIEW
)
session.readiness_state = derive_readiness_state(session, clarification_session)
session.recommended_action = derive_recommended_action(session, clarification_session)
session.current_phase = SessionPhase.CLARIFICATION if clarification_session.current_question_id else SessionPhase.REVIEW
self.repository.bump_session_version(session)
self.repository.db.commit()
self.repository.db.refresh(session)
logger.reflect(
"Clarification answer recorded and session advanced",
extra={
"session_id": session.session_id,
"question_id": question.question_id,
"next_question_id": clarification_session.current_question_id,
"readiness_state": session.readiness_state.value,
"remaining_count": clarification_session.remaining_count,
},
)
logger.reflect("Clarification answer recorded and session advanced", extra={"session_id": session.session_id, "question_id": question.question_id, "next_question_id": clarification_session.current_question_id, "readiness_state": session.readiness_state.value, "remaining_count": clarification_session.remaining_count})
return ClarificationStateResult(
clarification_session=clarification_session,
@@ -376,12 +266,11 @@ class ClarificationEngine:
# [/DEF:record_answer:Function]
# [DEF:summarize_progress:Function]
# @COMPLEXITY: 2
# @COMPLEXITY: 1
# @PURPOSE: Produce a compact progress summary for pause/resume and completion UX.
# @RELATION: [DEPENDS_ON] ->[ClarificationSession]
def summarize_progress(self, clarification_session: ClarificationSession) -> str:
resolved = self._count_resolved_questions(clarification_session)
remaining = self._count_remaining_questions(clarification_session)
resolved = count_resolved_questions(clarification_session)
remaining = count_remaining_questions(clarification_session)
return f"{resolved} resolved, {remaining} unresolved"
# [/DEF:summarize_progress:Function]
@@ -389,246 +278,25 @@ class ClarificationEngine:
# [DEF:_get_latest_clarification_session:Function]
# @COMPLEXITY: 2
# @PURPOSE: Select the latest clarification session for the current dataset review aggregate.
def _get_latest_clarification_session(
self,
session: DatasetReviewSession,
) -> Optional[ClarificationSession]:
def _get_latest_clarification_session(self, session: DatasetReviewSession) -> Optional[ClarificationSession]:
if not session.clarification_sessions:
return None
ordered_sessions = sorted(
session.clarification_sessions,
key=lambda item: (item.started_at, item.clarification_session_id),
reverse=True,
)
return ordered_sessions[0]
ordered = sorted(session.clarification_sessions, key=lambda item: (item.started_at, item.clarification_session_id), reverse=True)
return ordered[0]
# [/DEF:_get_latest_clarification_session:Function]
# [DEF:_find_question:Function]
# @COMPLEXITY: 2
# @COMPLEXITY: 1
# @PURPOSE: Resolve a clarification question from the active clarification aggregate.
def _find_question(
self,
clarification_session: ClarificationSession,
question_id: str,
) -> Optional[ClarificationQuestion]:
for question in clarification_session.questions:
if question.question_id == question_id:
return question
def _find_question(self, clarification_session: ClarificationSession, question_id: str) -> Optional[ClarificationQuestion]:
for q in clarification_session.questions:
if q.question_id == question_id:
return q
return None
# [/DEF:_find_question:Function]
# [DEF:_select_next_open_question:Function]
# @COMPLEXITY: 2
# @PURPOSE: Select the next unresolved question in deterministic priority order.
def _select_next_open_question(
self,
clarification_session: ClarificationSession,
) -> Optional[ClarificationQuestion]:
open_questions = [
question
for question in clarification_session.questions
if question.state == QuestionState.OPEN
]
if not open_questions:
return None
open_questions.sort(
key=lambda item: (-int(item.priority), item.created_at, item.question_id)
)
return open_questions[0]
# [/DEF:_select_next_open_question:Function]
# [DEF:_count_resolved_questions:Function]
# @COMPLEXITY: 2
# @PURPOSE: Count questions whose answers fully resolved the ambiguity.
def _count_resolved_questions(
self, clarification_session: ClarificationSession
) -> int:
return sum(
1
for question in clarification_session.questions
if question.state == QuestionState.ANSWERED
)
# [/DEF:_count_resolved_questions:Function]
# [DEF:_count_remaining_questions:Function]
# @COMPLEXITY: 2
# @PURPOSE: Count questions still unresolved or deferred after clarification interaction.
def _count_remaining_questions(
self, clarification_session: ClarificationSession
) -> int:
return sum(
1
for question in clarification_session.questions
if question.state
in {QuestionState.OPEN, QuestionState.SKIPPED, QuestionState.EXPERT_REVIEW}
)
# [/DEF:_count_remaining_questions:Function]
# [DEF:_normalize_answer_value:Function]
# @COMPLEXITY: 2
# @PURPOSE: Validate and normalize answer payload based on answer kind and active question options.
def _normalize_answer_value(
self,
answer_kind: AnswerKind,
answer_value: Optional[str],
question: ClarificationQuestion,
) -> Optional[str]:
normalized_answer_value = (
str(answer_value).strip() if answer_value is not None else None
)
if (
answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}
and not normalized_answer_value
):
raise ValueError(
"answer_value is required for selected or custom clarification answers"
)
if answer_kind == AnswerKind.SELECTED:
allowed_values = {option.value for option in question.options}
if normalized_answer_value not in allowed_values:
raise ValueError(
"answer_value must match one of the current clarification options"
)
if answer_kind == AnswerKind.SKIPPED:
return normalized_answer_value or "skipped"
if answer_kind == AnswerKind.EXPERT_REVIEW:
return normalized_answer_value or "expert_review"
return normalized_answer_value
# [/DEF:_normalize_answer_value:Function]
# [DEF:_build_impact_summary:Function]
# @COMPLEXITY: 2
# @PURPOSE: Build a compact audit note describing how the clarification answer affects session state.
def _build_impact_summary(
self,
question: ClarificationQuestion,
answer_kind: AnswerKind,
answer_value: Optional[str],
) -> str:
if answer_kind == AnswerKind.SKIPPED:
return f"Clarification for {question.topic_ref} was skipped and remains unresolved."
if answer_kind == AnswerKind.EXPERT_REVIEW:
return f"Clarification for {question.topic_ref} was deferred for expert review."
return f"Clarification for {question.topic_ref} recorded as '{answer_value}'."
# [/DEF:_build_impact_summary:Function]
# [DEF:_upsert_clarification_finding:Function]
# @COMPLEXITY: 2
# @PURPOSE: Keep one finding per clarification topic aligned with answer outcome and unresolved visibility rules.
# @RELATION: [DEPENDS_ON] ->[ValidationFinding]
def _upsert_clarification_finding(
self,
session: DatasetReviewSession,
question: ClarificationQuestion,
answer_kind: AnswerKind,
answer_value: Optional[str],
) -> ValidationFinding:
caused_by_ref = f"clarification:{question.question_id}"
existing = next(
(
finding
for finding in session.findings
if finding.area == FindingArea.CLARIFICATION
and finding.caused_by_ref == caused_by_ref
),
None,
)
if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}:
resolution_state = ResolutionState.RESOLVED
resolved_at = datetime.utcnow()
message = f"Clarified '{question.topic_ref}' with answer '{answer_value}'."
elif answer_kind == AnswerKind.SKIPPED:
resolution_state = ResolutionState.SKIPPED
resolved_at = None
message = f"Clarification for '{question.topic_ref}' was skipped and still needs review."
else:
resolution_state = ResolutionState.EXPERT_REVIEW
resolved_at = None
message = (
f"Clarification for '{question.topic_ref}' requires expert review."
)
if existing is None:
existing = ValidationFinding(
finding_id=str(uuid.uuid4()),
session_id=session.session_id,
area=FindingArea.CLARIFICATION,
severity=FindingSeverity.WARNING,
code="CLARIFICATION_PENDING",
title="Clarification pending",
message=message,
resolution_state=resolution_state,
resolution_note=None,
caused_by_ref=caused_by_ref,
created_at=datetime.utcnow(),
resolved_at=resolved_at,
)
self.repository.db.add(existing)
session.findings.append(existing)
else:
existing.message = message
existing.resolution_state = resolution_state
existing.resolved_at = resolved_at
if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}:
existing.code = "CLARIFICATION_RESOLVED"
existing.title = "Clarification resolved"
elif answer_kind == AnswerKind.SKIPPED:
existing.code = "CLARIFICATION_SKIPPED"
existing.title = "Clarification skipped"
else:
existing.code = "CLARIFICATION_EXPERT_REVIEW"
existing.title = "Clarification requires expert review"
return existing
# [/DEF:_upsert_clarification_finding:Function]
# [DEF:_derive_readiness_state:Function]
# @COMPLEXITY: 2
# @PURPOSE: Recompute readiness after clarification mutation while preserving unresolved visibility semantics.
# @RELATION: [DEPENDS_ON] ->[ClarificationSession]
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
def _derive_readiness_state(self, session: DatasetReviewSession) -> ReadinessState:
clarification_session = self._get_latest_clarification_session(session)
if clarification_session is None:
return session.readiness_state
if clarification_session.current_question_id:
return ReadinessState.CLARIFICATION_ACTIVE
if clarification_session.remaining_count > 0:
return ReadinessState.CLARIFICATION_NEEDED
return ReadinessState.REVIEW_READY
# [/DEF:_derive_readiness_state:Function]
# [DEF:_derive_recommended_action:Function]
# @COMPLEXITY: 2
# @PURPOSE: Recompute next-action guidance after clarification mutations.
def _derive_recommended_action(
self, session: DatasetReviewSession
) -> RecommendedAction:
clarification_session = self._get_latest_clarification_session(session)
if clarification_session is None:
return session.recommended_action
if clarification_session.current_question_id:
return RecommendedAction.ANSWER_NEXT_QUESTION
if clarification_session.remaining_count > 0:
return RecommendedAction.START_CLARIFICATION
return RecommendedAction.REVIEW_DOCUMENTATION
# [/DEF:_derive_recommended_action:Function]
# [/DEF:ClarificationEngine:Class]

220
backend/src/services/dataset_review/clarification_pkg/_helpers.py Normal file
View File

@@ -0,0 +1,220 @@
# [DEF:ClarificationHelpers:Module]
# @COMPLEXITY: 3
# @PURPOSE: Pure helper functions for clarification engine — question selection, counting, normalization, finding upsert, and readiness derivation.
# @LAYER: Domain
# @RELATION: DEPENDS_ON -> [DatasetReviewModels]
from __future__ import annotations
import uuid
from datetime import datetime
from typing import List, Optional
from src.models.dataset_review import (
AnswerKind,
ClarificationAnswer,
ClarificationQuestion,
ClarificationSession,
DatasetReviewSession,
FindingArea,
FindingSeverity,
QuestionState,
ReadinessState,
RecommendedAction,
ResolutionState,
ValidationFinding,
)
# [DEF:select_next_open_question:Function]
# @COMPLEXITY: 2
# @PURPOSE: Select the next unresolved question in deterministic priority order.
def select_next_open_question(
clarification_session: ClarificationSession,
) -> Optional[ClarificationQuestion]:
open_questions = [
q for q in clarification_session.questions if q.state == QuestionState.OPEN
]
if not open_questions:
return None
open_questions.sort(key=lambda item: (-int(item.priority), item.created_at, item.question_id))
return open_questions[0]
# [/DEF:select_next_open_question:Function]
# [DEF:count_resolved_questions:Function]
# @COMPLEXITY: 1
# @PURPOSE: Count questions whose answers fully resolved the ambiguity.
def count_resolved_questions(clarification_session: ClarificationSession) -> int:
return sum(1 for q in clarification_session.questions if q.state == QuestionState.ANSWERED)
# [/DEF:count_resolved_questions:Function]
# [DEF:count_remaining_questions:Function]
# @COMPLEXITY: 1
# @PURPOSE: Count questions still unresolved or deferred after clarification interaction.
def count_remaining_questions(clarification_session: ClarificationSession) -> int:
return sum(
1
for q in clarification_session.questions
if q.state in {QuestionState.OPEN, QuestionState.SKIPPED, QuestionState.EXPERT_REVIEW}
)
# [/DEF:count_remaining_questions:Function]
# [DEF:normalize_answer_value:Function]
# @COMPLEXITY: 2
# @PURPOSE: Validate and normalize answer payload based on answer kind and active question options.
def normalize_answer_value(
answer_kind: AnswerKind,
answer_value: Optional[str],
question: ClarificationQuestion,
) -> Optional[str]:
normalized = str(answer_value).strip() if answer_value is not None else None
if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM} and not normalized:
raise ValueError("answer_value is required for selected or custom clarification answers")
if answer_kind == AnswerKind.SELECTED:
allowed_values = {option.value for option in question.options}
if normalized not in allowed_values:
raise ValueError("answer_value must match one of the current clarification options")
if answer_kind == AnswerKind.SKIPPED:
return normalized or "skipped"
if answer_kind == AnswerKind.EXPERT_REVIEW:
return normalized or "expert_review"
return normalized
# [/DEF:normalize_answer_value:Function]
# [DEF:build_impact_summary:Function]
# @COMPLEXITY: 1
# @PURPOSE: Build a compact audit note describing how the clarification answer affects session state.
def build_impact_summary(
question: ClarificationQuestion,
answer_kind: AnswerKind,
answer_value: Optional[str],
) -> str:
if answer_kind == AnswerKind.SKIPPED:
return f"Clarification for {question.topic_ref} was skipped and remains unresolved."
if answer_kind == AnswerKind.EXPERT_REVIEW:
return f"Clarification for {question.topic_ref} was deferred for expert review."
return f"Clarification for {question.topic_ref} recorded as '{answer_value}'."
# [/DEF:build_impact_summary:Function]
# [DEF:upsert_clarification_finding:Function]
# @COMPLEXITY: 3
# @PURPOSE: Keep one finding per clarification topic aligned with answer outcome and unresolved visibility rules.
# @RELATION: DEPENDS_ON -> [ValidationFinding]
def upsert_clarification_finding(
session: DatasetReviewSession,
question: ClarificationQuestion,
answer_kind: AnswerKind,
answer_value: Optional[str],
db_session,
) -> Optional[ValidationFinding]:
caused_by_ref = f"clarification:{question.question_id}"
existing = next(
(f for f in session.findings if f.area == FindingArea.CLARIFICATION and f.caused_by_ref == caused_by_ref),
None,
)
if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}:
resolution_state = ResolutionState.RESOLVED
resolved_at = datetime.utcnow()
message = f"Clarified '{question.topic_ref}' with answer '{answer_value}'."
elif answer_kind == AnswerKind.SKIPPED:
resolution_state = ResolutionState.SKIPPED
resolved_at = None
message = f"Clarification for '{question.topic_ref}' was skipped and still needs review."
else:
resolution_state = ResolutionState.EXPERT_REVIEW
resolved_at = None
message = f"Clarification for '{question.topic_ref}' requires expert review."
if existing is None:
existing = ValidationFinding(
finding_id=str(uuid.uuid4()),
session_id=session.session_id,
area=FindingArea.CLARIFICATION,
severity=FindingSeverity.WARNING,
code="CLARIFICATION_PENDING",
title="Clarification pending",
message=message,
resolution_state=resolution_state,
resolution_note=None,
caused_by_ref=caused_by_ref,
created_at=datetime.utcnow(),
resolved_at=resolved_at,
)
db_session.add(existing)
session.findings.append(existing)
else:
existing.message = message
existing.resolution_state = resolution_state
existing.resolved_at = resolved_at
if answer_kind in {AnswerKind.SELECTED, AnswerKind.CUSTOM}:
existing.code = "CLARIFICATION_RESOLVED"
existing.title = "Clarification resolved"
elif answer_kind == AnswerKind.SKIPPED:
existing.code = "CLARIFICATION_SKIPPED"
existing.title = "Clarification skipped"
else:
existing.code = "CLARIFICATION_EXPERT_REVIEW"
existing.title = "Clarification requires expert review"
return existing
# [/DEF:upsert_clarification_finding:Function]
# [DEF:derive_readiness_state:Function]
# @COMPLEXITY: 2
# @PURPOSE: Recompute readiness after clarification mutation while preserving unresolved visibility semantics.
def derive_readiness_state(
session: DatasetReviewSession,
clarification_session: Optional[ClarificationSession],
) -> ReadinessState:
if clarification_session is None:
return session.readiness_state
if clarification_session.current_question_id:
return ReadinessState.CLARIFICATION_ACTIVE
if clarification_session.remaining_count > 0:
return ReadinessState.CLARIFICATION_NEEDED
return ReadinessState.REVIEW_READY
# [/DEF:derive_readiness_state:Function]
# [DEF:derive_recommended_action:Function]
# @COMPLEXITY: 2
# @PURPOSE: Recompute next-action guidance after clarification mutations.
def derive_recommended_action(
session: DatasetReviewSession,
clarification_session: Optional[ClarificationSession],
) -> RecommendedAction:
if clarification_session is None:
return session.recommended_action
if clarification_session.current_question_id:
return RecommendedAction.ANSWER_NEXT_QUESTION
if clarification_session.remaining_count > 0:
return RecommendedAction.START_CLARIFICATION
return RecommendedAction.REVIEW_DOCUMENTATION
# [/DEF:derive_recommended_action:Function]
# [/DEF:ClarificationHelpers:Module]

784
backend/src/services/dataset_review/orchestrator.py
View File

File diff suppressed because it is too large Load Diff

102
backend/src/services/dataset_review/orchestrator_pkg/_commands.py Normal file
View File

@@ -0,0 +1,102 @@
# [DEF:OrchestratorCommands:Module]
# @COMPLEXITY: 2
# @PURPOSE: Typed command and result dataclasses for dataset review orchestration boundary.
# @LAYER: Domain
# @RELATION: DEPENDS_ON -> [DatasetReviewModels]
# @RELATION: DEPENDS_ON -> [SupersetContextExtractor]
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from src.models.auth import User
from src.models.dataset_review import (
CompiledPreview,
DatasetReviewSession,
DatasetRunContext,
ValidationFinding,
)
from src.core.utils.superset_context_extractor import SupersetParsedContext
# [DEF:StartSessionCommand:Class]
# @COMPLEXITY: 2
# @PURPOSE: Typed input contract for starting a dataset review session.
@dataclass
class StartSessionCommand:
user: User
environment_id: str
source_kind: str
source_input: str
# [/DEF:StartSessionCommand:Class]
# [DEF:StartSessionResult:Class]
# @COMPLEXITY: 2
# @PURPOSE: Session-start result carrying the persisted session and intake recovery metadata.
@dataclass
class StartSessionResult:
session: DatasetReviewSession
parsed_context: Optional[SupersetParsedContext] = None
findings: List[ValidationFinding] = field(default_factory=list)
# [/DEF:StartSessionResult:Class]
# [DEF:PreparePreviewCommand:Class]
# @COMPLEXITY: 2
# @PURPOSE: Typed input contract for compiling one Superset-backed session preview.
@dataclass
class PreparePreviewCommand:
user: User
session_id: str
expected_version: Optional[int] = None
# [/DEF:PreparePreviewCommand:Class]
# [DEF:PreparePreviewResult:Class]
# @COMPLEXITY: 2
# @PURPOSE: Result contract for one persisted compiled preview attempt.
@dataclass
class PreparePreviewResult:
session: DatasetReviewSession
preview: CompiledPreview
blocked_reasons: List[str] = field(default_factory=list)
# [/DEF:PreparePreviewResult:Class]
# [DEF:LaunchDatasetCommand:Class]
# @COMPLEXITY: 2
# @PURPOSE: Typed input contract for launching one dataset-review session into SQL Lab.
@dataclass
class LaunchDatasetCommand:
user: User
session_id: str
expected_version: Optional[int] = None
# [/DEF:LaunchDatasetCommand:Class]
# [DEF:LaunchDatasetResult:Class]
# @COMPLEXITY: 2
# @PURPOSE: Launch result carrying immutable run context and any gate blockers.
@dataclass
class LaunchDatasetResult:
session: DatasetReviewSession
run_context: DatasetRunContext
blocked_reasons: List[str] = field(default_factory=list)
# [/DEF:LaunchDatasetResult:Class]
# [/DEF:OrchestratorCommands:Module]

356
backend/src/services/dataset_review/orchestrator_pkg/_helpers.py Normal file
View File

@@ -0,0 +1,356 @@
# [DEF:OrchestratorHelpers:Module]
# @COMPLEXITY: 4
# @PURPOSE: Pure helper methods extracted from DatasetReviewOrchestrator for INV_7 compliance — snapshot, blockers, fingerprint, recovery bootstrap.
# @LAYER: Domain
# @RELATION: DEPENDS_ON -> [DatasetReviewModels]
# @RELATION: DEPENDS_ON -> [SupersetContextExtractor]
# @PRE: Caller provides a loaded session aggregate with hydrated child collections.
# @POST: Helper results are deterministic and do not mutate persistence directly.
from __future__ import annotations
import hashlib
import json
from datetime import datetime
from typing import Any, Dict, List, Optional, cast
from src.core.logger import belief_scope, logger
from src.models.dataset_review import (
ApprovalState,
CompiledPreview,
ConfidenceState,
DatasetProfile,
DatasetReviewSession,
ExecutionMapping,
FilterConfidenceState,
FilterRecoveryStatus,
FilterSource,
FindingArea,
FindingSeverity,
ImportedFilter,
MappingMethod,
MappingStatus,
PreviewStatus,
ResolutionState,
TemplateVariable,
ValidationFinding,
VariableKind,
BusinessSummarySource,
)
logger = cast(Any, logger)
# [DEF:parse_dataset_selection:Function]
# @COMPLEXITY: 2
# @PURPOSE: Normalize dataset-selection payload into canonical session references.
def parse_dataset_selection(source_input: str) -> tuple[str, Optional[int]]:
normalized = str(source_input or "").strip()
if not normalized:
raise ValueError("dataset selection input must be non-empty")
if normalized.isdigit():
dataset_id = int(normalized)
return f"dataset:{dataset_id}", dataset_id
if normalized.startswith("dataset:"):
suffix = normalized.split(":", 1)[1].strip()
if suffix.isdigit():
return normalized, int(suffix)
return normalized, None
return normalized, None
# [/DEF:parse_dataset_selection:Function]
# [DEF:build_initial_profile:Function]
# @COMPLEXITY: 2
# @PURPOSE: Create the first profile snapshot so exports and detail views remain usable immediately after intake.
def build_initial_profile(
session_id: str,
parsed_context: Optional[Any],
dataset_ref: str,
) -> DatasetProfile:
dataset_name = (
dataset_ref.split(".")[-1] if dataset_ref else "Unresolved dataset"
)
business_summary = (
f"Review session initialized for {dataset_ref}."
if dataset_ref
else "Review session initialized with unresolved dataset context."
)
confidence_state = (
ConfidenceState.MIXED
if parsed_context and getattr(parsed_context, "partial_recovery", False)
else ConfidenceState.MOSTLY_CONFIRMED
)
return DatasetProfile(
session_id=session_id,
dataset_name=dataset_name or "Unresolved dataset",
schema_name=dataset_ref.split(".")[0] if "." in dataset_ref else None,
business_summary=business_summary,
business_summary_source=BusinessSummarySource.IMPORTED,
description="Initial review profile created from source intake.",
dataset_type="unknown",
is_sqllab_view=False,
completeness_score=0.25,
confidence_state=confidence_state,
has_blocking_findings=False,
has_warning_findings=bool(
parsed_context and getattr(parsed_context, "partial_recovery", False)
),
manual_summary_locked=False,
)
# [/DEF:build_initial_profile:Function]
# [DEF:build_partial_recovery_findings:Function]
# @COMPLEXITY: 3
# @PURPOSE: Project partial Superset intake recovery into explicit findings without blocking session usability.
# @PRE: parsed_context.partial_recovery is true.
# @POST: Returns warning-level findings that preserve usable but incomplete state.
def build_partial_recovery_findings(parsed_context: Any) -> List[ValidationFinding]:
findings: List[ValidationFinding] = []
for unresolved_ref in getattr(parsed_context, "unresolved_references", []):
findings.append(
ValidationFinding(
area=FindingArea.SOURCE_INTAKE,
severity=FindingSeverity.WARNING,
code="PARTIAL_SUPERSET_RECOVERY",
title="Superset context recovered partially",
message=(
"Session remains usable, but some Superset context requires review: "
f"{unresolved_ref.replace('_', ' ')}."
),
resolution_state=ResolutionState.OPEN,
caused_by_ref=unresolved_ref,
)
)
return findings
# [/DEF:build_partial_recovery_findings:Function]
# [DEF:extract_effective_filter_value:Function]
# @COMPLEXITY: 2
# @PURPOSE: Separate normalized filter payload metadata from the user-facing effective filter value.
def extract_effective_filter_value(
normalized_value: Any, raw_value: Any
) -> Any:
if isinstance(normalized_value, dict) and (
"filter_clauses" in normalized_value
or "extra_form_data" in normalized_value
):
return raw_value
return normalized_value if normalized_value is not None else raw_value
# [/DEF:extract_effective_filter_value:Function]
# [DEF:build_execution_snapshot:Function]
# @COMPLEXITY: 4
# @PURPOSE: Build effective filters, template params, approvals, and fingerprint for preview and launch gating.
# @PRE: Session aggregate includes imported filters, template variables, and current execution mappings.
# @POST: Returns deterministic execution snapshot for current session state without mutating persistence.
def build_execution_snapshot(session: DatasetReviewSession) -> Dict[str, Any]:
session_record = cast(Any, session)
filter_lookup = {
item.filter_id: item for item in session_record.imported_filters
}
variable_lookup = {
item.variable_id: item for item in session_record.template_variables
}
effective_filters: List[Dict[str, Any]] = []
template_params: Dict[str, Any] = {}
approved_mapping_ids: List[str] = []
open_warning_refs: List[str] = []
preview_blockers: List[str] = []
mapped_filter_ids: set[str] = set()
for mapping in session_record.execution_mappings:
imported_filter = filter_lookup.get(mapping.filter_id)
template_variable = variable_lookup.get(mapping.variable_id)
if imported_filter is None:
preview_blockers.append(f"mapping:{mapping.mapping_id}:missing_filter")
continue
if template_variable is None:
preview_blockers.append(f"mapping:{mapping.mapping_id}:missing_variable")
continue
effective_value = mapping.effective_value
if effective_value is None:
effective_value = extract_effective_filter_value(
imported_filter.normalized_value, imported_filter.raw_value,
)
if effective_value is None:
effective_value = template_variable.default_value
if effective_value is None and template_variable.is_required:
preview_blockers.append(
f"variable:{template_variable.variable_name}:missing_required_value"
)
continue
mapped_filter_ids.add(imported_filter.filter_id)
if effective_value is not None:
mapped_filter_payload = {
"mapping_id": mapping.mapping_id,
"filter_id": imported_filter.filter_id,
"filter_name": imported_filter.filter_name,
"variable_id": template_variable.variable_id,
"variable_name": template_variable.variable_name,
"effective_value": effective_value,
"raw_input_value": mapping.raw_input_value,
}
if isinstance(imported_filter.normalized_value, dict):
mapped_filter_payload["display_name"] = imported_filter.display_name
mapped_filter_payload["normalized_filter_payload"] = (
imported_filter.normalized_value
)
effective_filters.append(mapped_filter_payload)
template_params[template_variable.variable_name] = effective_value
if mapping.approval_state == ApprovalState.APPROVED:
approved_mapping_ids.append(mapping.mapping_id)
if (
mapping.requires_explicit_approval
and mapping.approval_state != ApprovalState.APPROVED
):
open_warning_refs.append(mapping.mapping_id)
for imported_filter in session_record.imported_filters:
if imported_filter.filter_id in mapped_filter_ids:
continue
effective_value = extract_effective_filter_value(
imported_filter.normalized_value, imported_filter.raw_value,
)
if effective_value is None:
continue
effective_filters.append(
{
"filter_id": imported_filter.filter_id,
"filter_name": imported_filter.filter_name,
"display_name": imported_filter.display_name,
"effective_value": effective_value,
"raw_input_value": imported_filter.raw_value,
"normalized_filter_payload": imported_filter.normalized_value,
}
)
mapped_variable_ids = {
mapping.variable_id for mapping in session_record.execution_mappings
}
for variable in session_record.template_variables:
if variable.variable_id in mapped_variable_ids:
continue
if variable.default_value is not None:
template_params[variable.variable_name] = variable.default_value
continue
if variable.is_required:
preview_blockers.append(f"variable:{variable.variable_name}:unmapped")
semantic_decision_refs = [
field.field_id
for field in session.semantic_fields
if field.is_locked
or not field.needs_review
or field.provenance.value != "unresolved"
]
preview_fingerprint = compute_preview_fingerprint(
{
"dataset_id": session_record.dataset_id,
"template_params": template_params,
"effective_filters": effective_filters,
}
)
return {
"effective_filters": effective_filters,
"template_params": template_params,
"approved_mapping_ids": sorted(approved_mapping_ids),
"semantic_decision_refs": sorted(semantic_decision_refs),
"open_warning_refs": sorted(open_warning_refs),
"preview_blockers": sorted(set(preview_blockers)),
"preview_fingerprint": preview_fingerprint,
}
# [/DEF:build_execution_snapshot:Function]
# [DEF:build_launch_blockers:Function]
# @COMPLEXITY: 3
# @PURPOSE: Enforce launch gates from findings, approvals, and current preview truth.
# @PRE: execution_snapshot was computed from current session state.
# @POST: Returns explicit blocker codes for every unmet launch invariant.
def build_launch_blockers(
session: DatasetReviewSession,
execution_snapshot: Dict[str, Any],
preview: Optional[CompiledPreview],
) -> List[str]:
session_record = cast(Any, session)
blockers = list(execution_snapshot["preview_blockers"])
for finding in session_record.findings:
if (
finding.severity == FindingSeverity.BLOCKING
and finding.resolution_state
not in {ResolutionState.RESOLVED, ResolutionState.APPROVED}
):
blockers.append(f"finding:{finding.code}:blocking")
for mapping in session_record.execution_mappings:
if (
mapping.requires_explicit_approval
and mapping.approval_state != ApprovalState.APPROVED
):
blockers.append(f"mapping:{mapping.mapping_id}:approval_required")
if preview is None:
blockers.append("preview:missing")
else:
if preview.preview_status != PreviewStatus.READY:
blockers.append(f"preview:{preview.preview_status.value}")
if preview.preview_fingerprint != execution_snapshot["preview_fingerprint"]:
blockers.append("preview:fingerprint_mismatch")
return sorted(set(blockers))
# [/DEF:build_launch_blockers:Function]
# [DEF:get_latest_preview:Function]
# @COMPLEXITY: 2
# @PURPOSE: Resolve the current latest preview snapshot for one session aggregate.
def get_latest_preview(session: DatasetReviewSession) -> Optional[CompiledPreview]:
session_record = cast(Any, session)
if not session_record.previews:
return None
if session_record.last_preview_id:
for preview in session_record.previews:
if preview.preview_id == session_record.last_preview_id:
return preview
return sorted(
session_record.previews,
key=lambda item: (item.created_at or datetime.min, item.preview_id),
reverse=True,
)[0]
# [/DEF:get_latest_preview:Function]
# [DEF:compute_preview_fingerprint:Function]
# @COMPLEXITY: 1
# @PURPOSE: Produce deterministic execution fingerprint for preview truth and staleness checks.
def compute_preview_fingerprint(payload: Dict[str, Any]) -> str:
serialized = json.dumps(payload, sort_keys=True, default=str)
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
# [/DEF:compute_preview_fingerprint:Function]
# [/DEF:OrchestratorHelpers:Module]

202
backend/src/services/dataset_review/repositories/repository_pkg/_mutations.py Normal file
View File

@@ -0,0 +1,202 @@
# [DEF:SessionRepositoryMutations:Module]
# @COMPLEXITY: 4
# @PURPOSE: Persistence mutation operations for dataset review session aggregates — profile/findings, recovery state, preview, run context.
# @LAYER: Domain
# @RELATION: DEPENDS_ON -> [DatasetReviewModels]
# @RELATION: DEPENDS_ON -> [SessionEventLogger]
# @PRE: All mutations execute within authenticated request or task scope.
# @POST: Session aggregate writes preserve ownership and version semantics.
from __future__ import annotations
from datetime import datetime
from typing import Any, List, Optional, cast
from sqlalchemy.orm import Session
from src.core.logger import belief_scope, logger
from src.models.dataset_review import (
ClarificationQuestion,
ClarificationSession,
CompiledPreview,
DatasetProfile,
DatasetReviewSession,
DatasetRunContext,
ExecutionMapping,
ImportedFilter,
SemanticFieldEntry,
SessionCollaborator,
SessionEvent,
TemplateVariable,
ValidationFinding,
)
from src.services.dataset_review.event_logger import SessionEventLogger
logger = cast(Any, logger)
# [DEF:save_profile_and_findings:Function]
# @COMPLEXITY: 4
# @PURPOSE: Persist profile state and replace validation findings for an owned session in one transaction.
# @PRE: session_id belongs to user_id and the supplied profile/findings belong to the same aggregate scope.
# @POST: stored profile matches the current session and findings are replaced by the supplied collection.
# @SIDE_EFFECT: updates profile rows, deletes stale findings, inserts current findings, and commits the transaction.
def save_profile_and_findings(
db: Session,
event_logger: SessionEventLogger,
get_owned_session,
require_session_version,
commit_session_mutation,
session_id: str,
user_id: str,
profile: DatasetProfile,
findings: List[ValidationFinding],
expected_version: Optional[int] = None,
) -> DatasetReviewSession:
with belief_scope("save_profile_and_findings"):
session = get_owned_session(session_id, user_id)
if expected_version is not None:
require_session_version(session, expected_version)
logger.reason("Persisting dataset profile and replacing validation findings", extra={"session_id": session_id, "user_id": user_id, "has_profile": bool(profile), "findings_count": len(findings)})
if profile:
existing_profile = db.query(DatasetProfile).filter_by(session_id=session_id).first()
if existing_profile:
profile.profile_id = existing_profile.profile_id
db.merge(profile)
db.query(ValidationFinding).filter(ValidationFinding.session_id == session_id).delete()
for finding in findings:
cast(Any, finding).session_id = session_id
db.add(finding)
commit_session_mutation(session, expected_version=expected_version)
logger.reflect("Dataset profile and validation findings committed", extra={"session_id": session.session_id, "user_id": user_id, "findings_count": len(findings)})
from src.services.dataset_review.repositories.session_repository import DatasetReviewSessionRepository
return session
# [/DEF:save_profile_and_findings:Function]
# [DEF:save_recovery_state:Function]
# @COMPLEXITY: 4
# @PURPOSE: Persist imported filters, template variables, and initial execution mappings for one owned session.
# @PRE: session_id belongs to user_id.
# @POST: Recovery state persisted to database.
# @SIDE_EFFECT: Writes to database.
def save_recovery_state(
db: Session,
get_owned_session,
require_session_version,
commit_session_mutation,
load_session_detail_fn,
session_id: str,
user_id: str,
imported_filters: List[ImportedFilter],
template_variables: List[TemplateVariable],
execution_mappings: List[ExecutionMapping],
expected_version: Optional[int] = None,
) -> DatasetReviewSession:
with belief_scope("save_recovery_state"):
session = get_owned_session(session_id, user_id)
if expected_version is not None:
require_session_version(session, expected_version)
logger.reason("Persisting dataset review recovery bootstrap state", extra={"session_id": session_id, "user_id": user_id, "imported_filters_count": len(imported_filters), "template_variables_count": len(template_variables), "execution_mappings_count": len(execution_mappings)})
db.query(ExecutionMapping).filter(ExecutionMapping.session_id == session_id).delete()
db.query(TemplateVariable).filter(TemplateVariable.session_id == session_id).delete()
db.query(ImportedFilter).filter(ImportedFilter.session_id == session_id).delete()
for f in imported_filters:
cast(Any, f).session_id = session_id
db.add(f)
for tv in template_variables:
cast(Any, tv).session_id = session_id
db.add(tv)
db.flush()
for em in execution_mappings:
cast(Any, em).session_id = session_id
db.add(em)
commit_session_mutation(session, expected_version=expected_version)
logger.reflect("Dataset review recovery bootstrap state committed", extra={"session_id": session.session_id, "user_id": user_id})
return load_session_detail_fn(session_id, user_id)
# [/DEF:save_recovery_state:Function]
# [DEF:save_preview:Function]
# @COMPLEXITY: 3
# @PURPOSE: Persist a preview snapshot and mark prior session previews stale.
# @PRE: session_id belongs to user_id and preview is prepared for the same session aggregate.
# @POST: preview is persisted and the session points to the latest preview identifier.
# @SIDE_EFFECT: updates prior preview statuses, inserts a preview row, mutates the parent session, and commits.
def save_preview(
db: Session,
get_owned_session,
require_session_version,
commit_session_mutation,
session_id: str,
user_id: str,
preview: CompiledPreview,
expected_version: Optional[int] = None,
) -> CompiledPreview:
with belief_scope("save_preview"):
session = get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
require_session_version(session, expected_version)
logger.reason("Persisting compiled preview and staling previous preview snapshots", extra={"session_id": session_id, "user_id": user_id})
db.query(CompiledPreview).filter(CompiledPreview.session_id == session_id).update({"preview_status": "stale"})
db.add(preview)
db.flush()
session_record.last_preview_id = preview.preview_id
commit_session_mutation(session, refresh_targets=[preview], expected_version=expected_version)
logger.reflect("Compiled preview committed as latest session preview", extra={"session_id": session.session_id, "preview_id": preview.preview_id})
return preview
# [/DEF:save_preview:Function]
# [DEF:save_run_context:Function]
# @COMPLEXITY: 3
# @PURPOSE: Persist an immutable launch audit snapshot for an owned session.
# @PRE: session_id belongs to user_id and run_context targets the same aggregate.
# @POST: run context is persisted and linked as the latest launch snapshot for the session.
# @SIDE_EFFECT: inserts a run-context row, mutates the parent session pointer, and commits.
def save_run_context(
db: Session,
get_owned_session,
require_session_version,
commit_session_mutation,
session_id: str,
user_id: str,
run_context: DatasetRunContext,
expected_version: Optional[int] = None,
) -> DatasetRunContext:
with belief_scope("save_run_context"):
session = get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
require_session_version(session, expected_version)
logger.reason("Persisting dataset run context audit snapshot", extra={"session_id": session_id, "user_id": user_id})
db.add(run_context)
db.flush()
session_record.last_run_context_id = run_context.run_context_id
commit_session_mutation(session, refresh_targets=[run_context], expected_version=expected_version)
logger.reflect("Dataset run context committed as latest launch snapshot", extra={"session_id": session.session_id, "run_context_id": run_context.run_context_id})
return run_context
# [/DEF:save_run_context:Function]
# [/DEF:SessionRepositoryMutations:Module]

524
backend/src/services/dataset_review/repositories/session_repository.py
View File

@@ -2,15 +2,18 @@
# @COMPLEXITY: 5
# @PURPOSE: Persist and retrieve dataset review session aggregates, including readiness, findings, semantic decisions, clarification state, previews, and run contexts.
# @LAYER: Domain
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [DatasetProfile]
# @RELATION: [DEPENDS_ON] -> [ValidationFinding]
# @RELATION: [DEPENDS_ON] -> [CompiledPreview]
# @RELATION: DEPENDS_ON -> [DatasetReviewSession]
# @RELATION: DEPENDS_ON -> [DatasetProfile]
# @RELATION: DEPENDS_ON -> [ValidationFinding]
# @RELATION: DEPENDS_ON -> [CompiledPreview]
# @RELATION: DISPATCHES -> [SessionRepositoryMutations:Module]
# @PRE: repository operations execute within authenticated request or task scope.
# @POST: session aggregate reads are structurally consistent and writes preserve ownership and version semantics.
# @SIDE_EFFECT: reads and writes SQLAlchemy-backed session aggregates.
# @DATA_CONTRACT: Input[SessionMutation] -> Output[PersistedSessionAggregate]
# @INVARIANT: answers, mapping approvals, preview artifacts, and launch snapshots are never attributed to the wrong user or session.
# @RATIONALE: Original 627-line file exceeded INV_7 (400-line module limit). Extracted mutation operations into _mutations sub-module.
# @REJECTED: Keeping all repository operations in one file because it exceeded the fractal limit.
from datetime import datetime
from typing import Any, Optional, List, cast
@@ -57,23 +60,17 @@ class DatasetReviewSessionVersionConflictError(ValueError):
# [DEF:DatasetReviewSessionRepository:Class]
# @COMPLEXITY: 4
# @PURPOSE: Enforce ownership-scoped persistence and retrieval for dataset review session aggregates.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [DatasetProfile]
# @RELATION: [DEPENDS_ON] -> [ValidationFinding]
# @RELATION: [DEPENDS_ON] -> [CompiledPreview]
# @RELATION: [DEPENDS_ON] -> [SessionEventLogger]
# @PRE: constructor receives a live SQLAlchemy session and callers provide authenticated user scope for guarded reads and writes.
# @RELATION: DEPENDS_ON -> [DatasetReviewSession]
# @RELATION: DEPENDS_ON -> [SessionEventLogger]
# @PRE: constructor receives a live SQLAlchemy session and callers provide authenticated user scope.
# @POST: repository methods return ownership-scoped aggregates or persisted child records without changing domain meaning.
# @SIDE_EFFECT: mutates and queries the persistence layer through the injected database session.
# @DATA_CONTRACT: Input[OwnedSessionQuery|SessionMutation] -> Output[PersistedSessionAggregate|PersistedChildRecord]
class DatasetReviewSessionRepository:
# [DEF:init_repo:Function]
# @COMPLEXITY: 4
# @COMPLEXITY: 2
# @PURPOSE: Bind one live SQLAlchemy session to the repository instance.
# @RELATION: DEPENDS_ON -> DatasetReviewSessionRepository; CALLS -> sqlalchemy
# @PRE: db_session is not None
# @POST: Repository instance initialized with valid session
# @SIDE_EFFECT: None - pure initialization
def __init__(self, db: Session):
self.db = db
self.event_logger = SessionEventLogger(db)
@@ -81,542 +78,205 @@ class DatasetReviewSessionRepository:
# [/DEF:init_repo:Function]
# [DEF:get_owned_session:Function]
# @COMPLEXITY: 4
# @PURPOSE: Resolve one owner-scoped dataset review session for mutation paths without leaking foreign-session state.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @COMPLEXITY: 3
# @PURPOSE: Resolve one owner-scoped dataset review session for mutation paths.
# @PRE: session_id and user_id are non-empty identifiers from the authenticated ownership scope.
# @POST: returns the owned session or raises a deterministic access error.
# @SIDE_EFFECT: reads one session row from the current database transaction.
# @DATA_CONTRACT: Input[OwnedSessionQuery] -> Output[DatasetReviewSession|ValueError]
def _get_owned_session(self, session_id: str, user_id: str) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.get_owned_session"):
logger.reason(
"Resolving owner-scoped dataset review session for mutation path",
extra={"session_id": session_id, "user_id": user_id},
)
logger.reason("Resolving owner-scoped dataset review session", extra={"session_id": session_id, "user_id": user_id})
session = (
self.db.query(DatasetReviewSession)
.filter(
DatasetReviewSession.session_id == session_id,
DatasetReviewSession.user_id == user_id,
)
.filter(DatasetReviewSession.session_id == session_id, DatasetReviewSession.user_id == user_id)
.first()
)
if not session:
logger.explore(
"Owner-scoped dataset review session lookup failed",
extra={"session_id": session_id, "user_id": user_id},
)
logger.explore("Owner-scoped dataset review session lookup failed", extra={"session_id": session_id, "user_id": user_id})
raise ValueError("Session not found or access denied")
logger.reflect(
"Owner-scoped dataset review session resolved",
extra={"session_id": session.session_id, "user_id": session.user_id},
)
logger.reflect("Owner-scoped dataset review session resolved", extra={"session_id": session.session_id})
return session
# [/DEF:get_owned_session:Function]
# [DEF:create_sess:Function]
# @COMPLEXITY: 4
# @COMPLEXITY: 3
# @PURPOSE: Persist an initial dataset review session shell.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @PRE: session is a new aggregate root bound to the current ownership scope.
# @POST: session is committed, refreshed, and returned with persisted identifiers.
# @SIDE_EFFECT: inserts a session row and commits the active transaction.
# @DATA_CONTRACT: Input[DatasetReviewSession] -> Output[DatasetReviewSession]
def create_session(self, session: DatasetReviewSession) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.create_session"):
logger.reason(
"Persisting dataset review session shell",
extra={
"user_id": session.user_id,
"environment_id": session.environment_id,
},
)
logger.reason("Persisting dataset review session shell", extra={"user_id": session.user_id, "environment_id": session.environment_id})
self.db.add(session)
self.db.commit()
self.db.refresh(session)
logger.reflect(
"Dataset review session shell persisted with stable identifier",
extra={"session_id": session.session_id, "user_id": session.user_id},
)
logger.reflect("Dataset review session shell persisted", extra={"session_id": session.session_id})
return session
# [/DEF:create_sess:Function]
# [DEF:require_session_version:Function]
# @COMPLEXITY: 4
# @COMPLEXITY: 3
# @PURPOSE: Enforce optimistic-lock version matching before a session mutation is persisted.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @PRE: session belongs to the current owner mutation scope and expected_version is the caller's last observed version.
# @POST: returns the same session when versions match; otherwise raises deterministic conflict error.
# @SIDE_EFFECT: none.
# @DATA_CONTRACT: Input[DatasetReviewSession,int] -> Output[DatasetReviewSession|DatasetReviewSessionVersionConflictError]
def require_session_version(
self, session: DatasetReviewSession, expected_version: int
) -> DatasetReviewSession:
def require_session_version(self, session: DatasetReviewSession, expected_version: int) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.require_session_version"):
session_record = cast(Any, session)
actual_version = int(getattr(session_record, "version", 0) or 0)
logger.reason(
"Checking optimistic-lock version for dataset review mutation",
extra={
"session_id": session.session_id,
"expected_version": expected_version,
"actual_version": actual_version,
},
)
actual_version = int(getattr(session, "version", 0) or 0)
logger.reason("Checking optimistic-lock version", extra={"session_id": session.session_id, "expected_version": expected_version, "actual_version": actual_version})
if actual_version != expected_version:
logger.explore(
"Rejected dataset review mutation due to stale session version",
extra={
"session_id": session.session_id,
"expected_version": expected_version,
"actual_version": actual_version,
},
)
raise DatasetReviewSessionVersionConflictError(
str(session_record.session_id), expected_version, actual_version
)
logger.reflect(
"Optimistic-lock version accepted for dataset review mutation",
extra={"session_id": session.session_id, "version": actual_version},
)
logger.explore("Rejected mutation due to stale session version", extra={"session_id": session.session_id, "expected_version": expected_version, "actual_version": actual_version})
raise DatasetReviewSessionVersionConflictError(str(session.session_id), expected_version, actual_version)
logger.reflect("Optimistic-lock version accepted", extra={"session_id": session.session_id, "version": actual_version})
return session
# [/DEF:require_session_version:Function]
# [DEF:bump_session_version:Function]
# @COMPLEXITY: 4
# @COMPLEXITY: 2
# @PURPOSE: Increment optimistic-lock version after a successful session mutation is assembled.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @PRE: session mutation has passed guards and will be committed in the current transaction.
# @POST: session version increments monotonically and last_activity_at reflects the mutation time.
# @SIDE_EFFECT: mutates the in-memory session aggregate before commit.
# @DATA_CONTRACT: Input[DatasetReviewSession] -> Output[int]
# @POST: session version increments monotonically.
def bump_session_version(self, session: DatasetReviewSession) -> int:
with belief_scope("DatasetReviewSessionRepository.bump_session_version"):
session_record = cast(Any, session)
next_version = int(getattr(session_record, "version", 0) or 0) + 1
session_record.version = next_version
session_record.last_activity_at = datetime.utcnow()
logger.reflect(
"Prepared incremented dataset review session version",
extra={"session_id": session.session_id, "version": next_version},
)
next_version = int(getattr(session, "version", 0) or 0) + 1
setattr(session, "version", next_version)
session.last_activity_at = datetime.utcnow()
logger.reflect("Prepared incremented session version", extra={"session_id": session.session_id, "version": next_version})
return next_version
# [/DEF:bump_session_version:Function]
# [DEF:commit_session_mutation:Function]
# @COMPLEXITY: 4
# @PURPOSE: Commit one prepared dataset review session mutation and translate stale writes into deterministic optimistic-lock conflicts.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @PRE: session mutation has already been assembled in the current SQLAlchemy transaction.
# @PURPOSE: Commit one prepared session mutation and translate stale writes into deterministic conflicts.
# @POST: session mutation is committed with one version increment or a deterministic conflict error is raised.
# @SIDE_EFFECT: increments session version, commits the transaction, refreshes ORM rows, or rolls back failed stale writes.
# @DATA_CONTRACT: Input[DatasetReviewSession,List[Any]|None,int|None] -> Output[DatasetReviewSession|DatasetReviewSessionVersionConflictError]
def commit_session_mutation(
self,
session: DatasetReviewSession,
*,
refresh_targets: Optional[List[Any]] = None,
expected_version: Optional[int] = None,
self, session: DatasetReviewSession, *, refresh_targets: Optional[List[Any]] = None, expected_version: Optional[int] = None,
) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.commit_session_mutation"):
session_record = cast(Any, session)
observed_version = int(
expected_version
if expected_version is not None
else getattr(session_record, "version", 0) or 0
)
logger.reason(
"Committing dataset review session mutation with optimistic lock",
extra={
"session_id": session.session_id,
"observed_version": observed_version,
"refresh_count": len(refresh_targets or []),
},
)
observed_version = int(expected_version if expected_version is not None else getattr(session, "version", 0) or 0)
logger.reason("Committing session mutation with optimistic lock", extra={"session_id": session.session_id, "observed_version": observed_version})
self.bump_session_version(session)
try:
self.db.commit()
except StaleDataError as exc:
self.db.rollback()
actual_version_row = (
self.db.query(DatasetReviewSession.version)
.filter(DatasetReviewSession.session_id == session.session_id)
.first()
)
actual_version = (
int(actual_version_row[0] or 0) if actual_version_row else 0
)
logger.explore(
"Dataset review session commit rejected by optimistic lock",
extra={
"session_id": session.session_id,
"expected_version": observed_version,
"actual_version": actual_version,
},
)
raise DatasetReviewSessionVersionConflictError(
session.session_id,
observed_version,
actual_version,
) from exc
actual_version_row = self.db.query(DatasetReviewSession.version).filter(DatasetReviewSession.session_id == session.session_id).first()
actual_version = int(actual_version_row[0] or 0) if actual_version_row else 0
logger.explore("Session commit rejected by optimistic lock", extra={"session_id": session.session_id, "expected_version": observed_version, "actual_version": actual_version})
raise DatasetReviewSessionVersionConflictError(session.session_id, observed_version, actual_version) from exc
self.db.refresh(session)
for target in refresh_targets or []:
self.db.refresh(target)
logger.reflect(
"Dataset review session mutation committed",
extra={
"session_id": session.session_id,
"version": getattr(session, "version", None),
"refresh_count": len(refresh_targets or []),
},
)
logger.reflect("Session mutation committed", extra={"session_id": session.session_id, "version": getattr(session, "version", None)})
return session
# [/DEF:commit_session_mutation:Function]
# [DEF:load_detail:Function]
# @COMPLEXITY: 4
# @COMPLEXITY: 3
# @PURPOSE: Return the full session aggregate for API and frontend resume flows.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [SessionCollaborator]
# @PRE: session_id is a valid UUID; db_session is active
# @POST: Returns SessionDetail with all fields populated
# @SIDE_EFFECT: Read-only database operation
def load_session_detail(
self, session_id: str, user_id: str
) -> Optional[DatasetReviewSession]:
# @POST: Returns SessionDetail with all fields populated or None.
def load_session_detail(self, session_id: str, user_id: str) -> Optional[DatasetReviewSession]:
with belief_scope("DatasetReviewSessionRepository.load_session_detail"):
logger.reason(
"Loading dataset review session detail for owner-or-collaborator scope",
extra={"session_id": session_id, "user_id": user_id},
)
logger.reason("Loading dataset review session detail", extra={"session_id": session_id, "user_id": user_id})
session = (
self.db.query(DatasetReviewSession)
.outerjoin(
SessionCollaborator,
DatasetReviewSession.session_id == SessionCollaborator.session_id,
)
.outerjoin(SessionCollaborator, DatasetReviewSession.session_id == SessionCollaborator.session_id)
.options(
joinedload(DatasetReviewSession.profile),
joinedload(DatasetReviewSession.findings),
joinedload(DatasetReviewSession.collaborators),
joinedload(DatasetReviewSession.semantic_sources),
joinedload(DatasetReviewSession.semantic_fields).joinedload(
SemanticFieldEntry.candidates
),
joinedload(DatasetReviewSession.semantic_fields).joinedload(SemanticFieldEntry.candidates),
joinedload(DatasetReviewSession.imported_filters),
joinedload(DatasetReviewSession.template_variables),
joinedload(DatasetReviewSession.execution_mappings),
joinedload(DatasetReviewSession.clarification_sessions)
.joinedload(ClarificationSession.questions)
.joinedload(ClarificationQuestion.options),
joinedload(DatasetReviewSession.clarification_sessions)
.joinedload(ClarificationSession.questions)
.joinedload(ClarificationQuestion.answer),
joinedload(DatasetReviewSession.clarification_sessions).joinedload(ClarificationSession.questions).joinedload(ClarificationQuestion.options),
joinedload(DatasetReviewSession.clarification_sessions).joinedload(ClarificationSession.questions).joinedload(ClarificationQuestion.answer),
joinedload(DatasetReviewSession.previews),
joinedload(DatasetReviewSession.run_contexts),
joinedload(DatasetReviewSession.events),
)
.filter(DatasetReviewSession.session_id == session_id)
.filter(
or_(
DatasetReviewSession.user_id == user_id,
SessionCollaborator.user_id == user_id,
)
)
.filter(or_(DatasetReviewSession.user_id == user_id, SessionCollaborator.user_id == user_id))
.first()
)
logger.reflect(
"Dataset review session detail lookup completed",
extra={
"session_id": session_id,
"user_id": user_id,
"found": bool(session),
},
)
logger.reflect("Session detail lookup completed", extra={"session_id": session_id, "found": bool(session)})
return session
# [/DEF:load_detail:Function]
# [DEF:save_prof_find:Function]
# [DEF:save_profile_and_findings:Function]
# @COMPLEXITY: 4
# @PURPOSE: Persist profile state and replace validation findings for an owned session in one transaction.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [DatasetProfile]
# @RELATION: [DEPENDS_ON] -> [ValidationFinding]
# @PRE: session_id belongs to user_id and the supplied profile/findings belong to the same aggregate scope.
# @POST: stored profile matches the current session and findings are replaced by the supplied collection.
# @SIDE_EFFECT: updates profile rows, deletes stale findings, inserts current findings, and commits the transaction.
# @DATA_CONTRACT: Input[ProfileAndFindingsMutation] -> Output[DatasetReviewSession]
# @PURPOSE: Persist profile state and replace validation findings for an owned session.
# @POST: stored profile matches the current session and findings are replaced.
def save_profile_and_findings(
self,
session_id: str,
user_id: str,
profile: DatasetProfile,
findings: List[ValidationFinding],
expected_version: Optional[int] = None,
self, session_id: str, user_id: str, profile: DatasetProfile, findings: List[ValidationFinding], expected_version: Optional[int] = None,
) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.save_profile_and_findings"):
session = self._get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
self.require_session_version(session, expected_version)
logger.reason(
"Persisting dataset profile and replacing validation findings",
extra={
"session_id": session_id,
"user_id": user_id,
"has_profile": bool(profile),
"findings_count": len(findings),
"expected_version": expected_version,
},
)
from src.services.dataset_review.repositories.repository_pkg._mutations import save_profile_and_findings as _save
return _save(
self.db, self.event_logger, self._get_owned_session, self.require_session_version,
self.commit_session_mutation, session_id, user_id, profile, findings, expected_version,
)
if profile:
existing_profile = (
self.db.query(DatasetProfile)
.filter_by(session_id=session_id)
.first()
)
if existing_profile:
profile.profile_id = existing_profile.profile_id
self.db.merge(profile)
self.db.query(ValidationFinding).filter(
ValidationFinding.session_id == session_id
).delete()
for finding in findings:
finding_record = cast(Any, finding)
finding_record.session_id = session_id
self.db.add(finding)
self.commit_session_mutation(session, expected_version=expected_version)
logger.reflect(
"Dataset profile and validation findings committed",
extra={
"session_id": session.session_id,
"version": session_record.version,
"user_id": user_id,
"findings_count": len(findings),
},
)
return self.load_session_detail(session_id, user_id)
# [/DEF:save_prof_find:Function]
# [/DEF:save_profile_and_findings:Function]
# [DEF:save_recovery_state:Function]
# @COMPLEXITY: 4
# @PURPOSE: Persist imported filters, template variables, and initial execution mappings for one owned session.
# @RELATION: [DEPENDS_ON] -> [ImportedFilter]
# @RELATION: [DEPENDS_ON] -> [TemplateVariable]
# @RELATION: [DEPENDS_ON] -> [ExecutionMapping]
# @PRE: session_id is a valid UUID; recovery_state is a valid dict
# @POST: Recovery state persisted to database
# @SIDE_EFFECT: Writes to database
# @COMPLEXITY: 3
# @PURPOSE: Persist imported filters, template variables, and initial execution mappings.
def save_recovery_state(
self,
session_id: str,
user_id: str,
imported_filters: List[ImportedFilter],
template_variables: List[TemplateVariable],
execution_mappings: List[ExecutionMapping],
self, session_id: str, user_id: str, imported_filters: List[ImportedFilter],
template_variables: List[TemplateVariable], execution_mappings: List[ExecutionMapping],
expected_version: Optional[int] = None,
) -> DatasetReviewSession:
with belief_scope("DatasetReviewSessionRepository.save_recovery_state"):
session = self._get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
self.require_session_version(session, expected_version)
logger.reason(
"Persisting dataset review recovery bootstrap state",
extra={
"session_id": session_id,
"user_id": user_id,
"imported_filters_count": len(imported_filters),
"template_variables_count": len(template_variables),
"execution_mappings_count": len(execution_mappings),
"expected_version": expected_version,
},
)
self.db.query(ExecutionMapping).filter(
ExecutionMapping.session_id == session_id
).delete()
self.db.query(TemplateVariable).filter(
TemplateVariable.session_id == session_id
).delete()
self.db.query(ImportedFilter).filter(
ImportedFilter.session_id == session_id
).delete()
for imported_filter in imported_filters:
imported_filter_record = cast(Any, imported_filter)
imported_filter_record.session_id = session_id
self.db.add(imported_filter)
for template_variable in template_variables:
template_variable_record = cast(Any, template_variable)
template_variable_record.session_id = session_id
self.db.add(template_variable)
self.db.flush()
for execution_mapping in execution_mappings:
execution_mapping_record = cast(Any, execution_mapping)
execution_mapping_record.session_id = session_id
self.db.add(execution_mapping)
self.commit_session_mutation(session, expected_version=expected_version)
logger.reflect(
"Dataset review recovery bootstrap state committed",
extra={
"session_id": session.session_id,
"version": session_record.version,
"user_id": user_id,
"imported_filters_count": len(imported_filters),
"template_variables_count": len(template_variables),
"execution_mappings_count": len(execution_mappings),
},
)
return self.load_session_detail(session_id, user_id)
from src.services.dataset_review.repositories.repository_pkg._mutations import save_recovery_state as _save
return _save(
self.db, self._get_owned_session, self.require_session_version,
self.commit_session_mutation, self.load_session_detail,
session_id, user_id, imported_filters, template_variables, execution_mappings, expected_version,
)
# [/DEF:save_recovery_state:Function]
# [DEF:save_prev:Function]
# @COMPLEXITY: 4
# [DEF:save_preview:Function]
# @COMPLEXITY: 3
# @PURPOSE: Persist a preview snapshot and mark prior session previews stale.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [CompiledPreview]
# @PRE: session_id belongs to user_id and preview is prepared for the same session aggregate.
# @POST: preview is persisted and the session points to the latest preview identifier.
# @SIDE_EFFECT: updates prior preview statuses, inserts a preview row, mutates the parent session, and commits.
# @DATA_CONTRACT: Input[PreviewMutation] -> Output[CompiledPreview]
def save_preview(
self,
session_id: str,
user_id: str,
preview: CompiledPreview,
expected_version: Optional[int] = None,
self, session_id: str, user_id: str, preview: CompiledPreview, expected_version: Optional[int] = None,
) -> CompiledPreview:
with belief_scope("DatasetReviewSessionRepository.save_preview"):
session = self._get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
self.require_session_version(session, expected_version)
logger.reason(
"Persisting compiled preview and staling previous preview snapshots",
extra={
"session_id": session_id,
"user_id": user_id,
"expected_version": expected_version,
},
)
from src.services.dataset_review.repositories.repository_pkg._mutations import save_preview as _save
return _save(
self.db, self._get_owned_session, self.require_session_version,
self.commit_session_mutation, session_id, user_id, preview, expected_version,
)
self.db.query(CompiledPreview).filter(
CompiledPreview.session_id == session_id
).update({"preview_status": "stale"})
# [/DEF:save_preview:Function]
self.db.add(preview)
self.db.flush()
session_record.last_preview_id = preview.preview_id
self.commit_session_mutation(
session,
refresh_targets=[preview],
expected_version=expected_version,
)
logger.reflect(
"Compiled preview committed as latest session preview",
extra={
"session_id": session.session_id,
"version": session_record.version,
"preview_id": preview.preview_id,
"user_id": user_id,
},
)
return preview
# [/DEF:save_prev:Function]
# [DEF:save_run_ctx:Function]
# @COMPLEXITY: 4
# [DEF:save_run_context:Function]
# @COMPLEXITY: 3
# @PURPOSE: Persist an immutable launch audit snapshot for an owned session.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
# @RELATION: [DEPENDS_ON] -> [DatasetRunContext]
# @PRE: session_id belongs to user_id and run_context targets the same aggregate.
# @POST: run context is persisted and linked as the latest launch snapshot for the session.
# @SIDE_EFFECT: inserts a run-context row, mutates the parent session pointer, and commits.
# @DATA_CONTRACT: Input[RunContextMutation] -> Output[DatasetRunContext]
def save_run_context(
self,
session_id: str,
user_id: str,
run_context: DatasetRunContext,
expected_version: Optional[int] = None,
self, session_id: str, user_id: str, run_context: DatasetRunContext, expected_version: Optional[int] = None,
) -> DatasetRunContext:
with belief_scope("DatasetReviewSessionRepository.save_run_context"):
session = self._get_owned_session(session_id, user_id)
session_record = cast(Any, session)
if expected_version is not None:
self.require_session_version(session, expected_version)
logger.reason(
"Persisting dataset run context audit snapshot",
extra={
"session_id": session_id,
"user_id": user_id,
"expected_version": expected_version,
},
)
from src.services.dataset_review.repositories.repository_pkg._mutations import save_run_context as _save
return _save(
self.db, self._get_owned_session, self.require_session_version,
self.commit_session_mutation, session_id, user_id, run_context, expected_version,
)
self.db.add(run_context)
self.db.flush()
session_record.last_run_context_id = run_context.run_context_id
self.commit_session_mutation(
session,
refresh_targets=[run_context],
expected_version=expected_version,
)
logger.reflect(
"Dataset run context committed as latest launch snapshot",
extra={
"session_id": session.session_id,
"version": session_record.version,
"run_context_id": run_context.run_context_id,
"user_id": user_id,
},
)
return run_context
# [/DEF:save_run_ctx:Function]
# [/DEF:save_run_context:Function]
# [DEF:list_user_sess:Function]
# @COMPLEXITY: 2
# @PURPOSE: List review sessions owned by a specific user ordered by most recent update.
# @RELATION: [DEPENDS_ON] -> [DatasetReviewSession]
def list_sessions_for_user(self, user_id: str) -> List[DatasetReviewSession]:
with belief_scope("DatasetReviewSessionRepository.list_sessions_for_user"):
logger.reason(
"Listing dataset review sessions for owner scope",
extra={"user_id": user_id},
)
logger.reason("Listing dataset review sessions for owner scope", extra={"user_id": user_id})
sessions = (
self.db.query(DatasetReviewSession)
.filter(DatasetReviewSession.user_id == user_id)
.order_by(DatasetReviewSession.updated_at.desc())
.all()
)
logger.reflect(
"Dataset review session list assembled",
extra={"user_id": user_id, "session_count": len(sessions)},
)
logger.reflect("Session list assembled", extra={"user_id": user_id, "session_count": len(sessions)})
return sessions
# [/DEF:list_user_sess:Function]