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

skills + agents

Browse Source
This commit is contained in:
busya
2026-04-19 20:05:45 +03:00
parent 721caee611
commit 6a2741182b
39 changed files with 1155 additions and 1868 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -451,6 +451,7 @@ class DatasetReviewOrchestrator:
session.session_id,
command.user.id,
preview,
expected_version=command.expected_version,
)
session.current_phase = SessionPhase.PREVIEW
@@ -598,6 +599,7 @@ class DatasetReviewOrchestrator:
session.session_id,
command.user.id,
run_context,
expected_version=command.expected_version,
)
session.current_phase = SessionPhase.LAUNCH

103
backend/src/services/dataset_review/repositories/__tests__/test_session_repository.py
View File

@@ -1,6 +1,7 @@
import pytest
from sqlalchemy import create_engine, inspect, text
from sqlalchemy.orm import sessionmaker
from pathlib import Path
from src.core.database import _ensure_dataset_review_session_columns
from src.models.mapping import Base, Environment
from src.models.auth import User
@@ -352,6 +353,7 @@ def test_save_profile_and_findings(db_session):
assert updated_session.profile.dataset_name == "Test DS"
assert len(updated_session.findings) == 1
assert updated_session.findings[0].code == "ERR1"
assert updated_session.version == 1
# Verify removal of old findings
new_finding = ValidationFinding(
@@ -369,11 +371,112 @@ def test_save_profile_and_findings(db_session):
final_session = repo.load_session_detail(session.session_id, "user1")
assert len(final_session.findings) == 1
assert final_session.findings[0].code == "WARN1"
assert final_session.version == 2
# [/DEF:test_save_profile_and_findings:Function]
# [DEF:test_save_profile_and_findings_rejects_stale_concurrent_write:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
# @PURPOSE: Verify repository save path translates concurrent stale session writes into deterministic optimistic-lock conflicts.
def test_save_profile_and_findings_rejects_stale_concurrent_write(tmp_path: Path):
db_path = tmp_path / "dataset_review_session_repository.sqlite"
engine = create_engine(f"sqlite:///{db_path}")
Base.metadata.create_all(engine)
SessionFactory = sessionmaker(bind=engine)
seed_session = SessionFactory()
seed_session.add_all(
[
User(
id="user1",
username="testuser",
email="test@example.com",
password_hash="pw",
),
Environment(
id="env1",
name="Prod",
url="http://superset",
credentials_id="cred1",
),
]
)
seed_session.commit()
created_session = DatasetReviewSession(
user_id="user1",
environment_id="env1",
source_kind="superset_link",
source_input="http://link",
dataset_ref="dataset1",
)
seed_session.add(created_session)
seed_session.commit()
session_id = created_session.session_id
seed_session.close()
writer_a = SessionFactory()
writer_b = SessionFactory()
repo_a = DatasetReviewSessionRepository(writer_a)
repo_b = DatasetReviewSessionRepository(writer_b)
repo_a.save_profile_and_findings(
session_id,
"user1",
DatasetProfile(
session_id=session_id,
dataset_name="Writer A",
business_summary="Summary A",
business_summary_source=BusinessSummarySource.INFERRED,
confidence_state=ConfidenceState.UNRESOLVED,
),
[
ValidationFinding(
session_id=session_id,
area=FindingArea.SOURCE_INTAKE,
severity=FindingSeverity.WARNING,
code="WARN_A",
title="Warning A",
message="Writer A saved first",
)
],
)
with pytest.raises(DatasetReviewSessionVersionConflictError) as exc_info:
repo_b.save_profile_and_findings(
session_id,
"user1",
DatasetProfile(
session_id=session_id,
dataset_name="Writer B",
business_summary="Summary B",
business_summary_source=BusinessSummarySource.INFERRED,
confidence_state=ConfidenceState.UNRESOLVED,
),
[
ValidationFinding(
session_id=session_id,
area=FindingArea.DATASET_PROFILE,
severity=FindingSeverity.BLOCKING,
code="ERR_B",
title="Error B",
message="Writer B lost optimistic lock",
)
],
expected_version=0,
)
assert exc_info.value.expected_version == 0
assert exc_info.value.actual_version == 1
writer_a.close()
writer_b.close()
# [/DEF:test_save_profile_and_findings_rejects_stale_concurrent_write:Function]
# [DEF:test_save_run_context:Function]
# @RELATION: BINDS_TO -> SessionRepositoryTests
def test_save_run_context(db_session):

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

@@ -16,6 +16,7 @@ from datetime import datetime
from typing import Any, Optional, List, cast
from sqlalchemy import or_
from sqlalchemy.orm import Session, joinedload
from sqlalchemy.orm.exc import StaleDataError
from src.models.dataset_review import (
ClarificationQuestion,
ClarificationSession,
@@ -207,6 +208,78 @@ class DatasetReviewSessionRepository:
# [/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.
# @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,
) -> 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 []),
},
)
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
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 []),
},
)
return session
# [/DEF:commit_session_mutation:Function]
# [DEF:load_detail:Function]
# @COMPLEXITY: 4
# @PURPOSE: Return the full session aggregate for API and frontend resume flows.
@@ -287,10 +360,13 @@ class DatasetReviewSessionRepository:
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={
@@ -298,6 +374,7 @@ class DatasetReviewSessionRepository:
"user_id": user_id,
"has_profile": bool(profile),
"findings_count": len(findings),
"expected_version": expected_version,
},
)
@@ -320,8 +397,7 @@ class DatasetReviewSessionRepository:
finding_record.session_id = session_id
self.db.add(finding)
self.bump_session_version(session)
self.db.commit()
self.commit_session_mutation(session, expected_version=expected_version)
logger.reflect(
"Dataset profile and validation findings committed",
extra={
@@ -351,10 +427,13 @@ class DatasetReviewSessionRepository:
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={
@@ -363,6 +442,7 @@ class DatasetReviewSessionRepository:
"imported_filters_count": len(imported_filters),
"template_variables_count": len(template_variables),
"execution_mappings_count": len(execution_mappings),
"expected_version": expected_version,
},
)
@@ -393,8 +473,7 @@ class DatasetReviewSessionRepository:
execution_mapping_record.session_id = session_id
self.db.add(execution_mapping)
self.bump_session_version(session)
self.db.commit()
self.commit_session_mutation(session, expected_version=expected_version)
logger.reflect(
"Dataset review recovery bootstrap state committed",
extra={
@@ -420,14 +499,24 @@ class DatasetReviewSessionRepository:
# @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
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},
extra={
"session_id": session_id,
"user_id": user_id,
"expected_version": expected_version,
},
)
self.db.query(CompiledPreview).filter(
@@ -438,9 +527,11 @@ class DatasetReviewSessionRepository:
self.db.flush()
session_record.last_preview_id = preview.preview_id
self.bump_session_version(session)
self.db.commit()
self.db.refresh(preview)
self.commit_session_mutation(
session,
refresh_targets=[preview],
expected_version=expected_version,
)
logger.reflect(
"Compiled preview committed as latest session preview",
extra={
@@ -464,23 +555,35 @@ class DatasetReviewSessionRepository:
# @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
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},
extra={
"session_id": session_id,
"user_id": user_id,
"expected_version": expected_version,
},
)
self.db.add(run_context)
self.db.flush()
session_record.last_run_context_id = run_context.run_context_id
self.bump_session_version(session)
self.db.commit()
self.db.refresh(run_context)
self.commit_session_mutation(
session,
refresh_targets=[run_context],
expected_version=expected_version,
)
logger.reflect(
"Dataset run context committed as latest launch snapshot",
extra={