| """OpenEnv server-side environment for operational data cleaning tasks.""" |
|
|
| from __future__ import annotations |
|
|
| import copy |
| import random |
| from uuid import uuid4 |
|
|
| from openenv.core.env_server.interfaces import Environment |
| from openenv.core.env_server.types import EnvironmentMetadata |
|
|
| from cleanops_env.graders import build_table_summary, count_duplicate_groups, grade_tables |
| from cleanops_env.models import ( |
| ActionCostEntry, |
| DataCleaningAction, |
| DataCleaningObservation, |
| DataCleaningState, |
| DownstreamHealth, |
| DryRunFinding, |
| DryRunReport, |
| OperationDetail, |
| OperationSummary, |
| PendingReview, |
| ReviewResolution, |
| ReviewTarget, |
| RiskCard, |
| RewardBreakdown, |
| RowChange, |
| TableView, |
| ) |
| from cleanops_env.tasks import ( |
| ReviewCaseSpec, |
| TaskSpec, |
| apply_operation_to_tables, |
| clone_tables, |
| first_table_name, |
| get_task_spec, |
| list_task_ids, |
| normalize_whitespace, |
| sorted_rows, |
| ) |
|
|
| ACTION_COSTS: dict[str, float] = { |
| "inspect_table": 0.005, |
| "inspect_operation": 0.005, |
| "apply_operation:safe": 0.01, |
| "apply_operation:review": 0.015, |
| "apply_operation:destructive": 0.03, |
| "request_review": 0.025, |
| "run_sync_dry_run": 0.02, |
| "submit": 0.005, |
| } |
|
|
| ACTION_COST_DESCRIPTIONS: dict[str, str] = { |
| "inspect_table": "Low-cost inspection to understand current records.", |
| "inspect_operation": "Low-cost preview to inspect an operation before applying it.", |
| "apply_operation:safe": "Safe automated cleanup with low operational risk.", |
| "apply_operation:review": "Review-sensitive cleanup that should be used more deliberately.", |
| "apply_operation:destructive": "Destructive cleanup with higher business risk if applied incorrectly.", |
| "request_review": "Consumes limited human-review budget to resolve ambiguity safely.", |
| "run_sync_dry_run": "Runs a deterministic downstream system simulation before submit.", |
| "submit": "Low-cost finalization step after cleanup is complete.", |
| } |
|
|
|
|
| class CleanOpsEnvironment(Environment[DataCleaningAction, DataCleaningObservation, DataCleaningState]): |
| """A realistic data-cleaning workflow environment with deterministic graders.""" |
|
|
| SUPPORTS_CONCURRENT_SESSIONS = True |
|
|
| def __init__(self) -> None: |
| super().__init__() |
| self._task_order = list_task_ids() |
| self._task_spec = get_task_spec(self._task_order[0]) |
| self._grade = grade_tables(self._task_spec, self._task_spec.dirty_tables) |
| self._focus_table_name = first_table_name(self._task_spec) |
| self._focus_operation_detail: OperationDetail | None = None |
| self._done = False |
| self._initial_issue_count = max(1, len(self._grade.validation_issues)) |
| initial_tables = clone_tables(self._task_spec.dirty_tables) |
| initial_downstream_health = self._compute_downstream_health(self._task_spec, initial_tables, self._grade.validation_issues) |
| self._state = DataCleaningState( |
| episode_id=str(uuid4()), |
| step_count=0, |
| task_id=self._task_spec.task_id, |
| task_title=self._task_spec.title, |
| difficulty=self._task_spec.difficulty, |
| requested_seed=None, |
| max_steps=self._task_spec.max_steps, |
| review_budget_total=self._task_spec.review_budget, |
| review_budget_remaining=self._task_spec.review_budget, |
| submitted=False, |
| current_score=self._grade.score, |
| best_score=self._grade.score, |
| outstanding_issue_count=len(self._grade.validation_issues), |
| downstream_health=initial_downstream_health, |
| last_dry_run=None, |
| tables=initial_tables, |
| applied_operation_ids=[], |
| inspected_tables=[self._focus_table_name], |
| inspected_operations=[], |
| requested_review_ids=[], |
| pending_reviews=[], |
| resolved_reviews=[], |
| dry_run_targets=[], |
| recent_history=[], |
| ) |
|
|
| def reset( |
| self, |
| seed: int | None = None, |
| episode_id: str | None = None, |
| task_id: str | None = None, |
| **kwargs: object, |
| ) -> DataCleaningObservation: |
| del kwargs |
| selected_task_id = task_id or self._task_order[0] |
| self._task_spec = get_task_spec(selected_task_id) |
| normalized_seed = seed if seed is None else max(0, int(seed)) |
| self._focus_table_name = self._choose_initial_focus_table(self._task_spec, normalized_seed) |
| self._focus_operation_detail = None |
| self._done = False |
| self._grade = grade_tables(self._task_spec, self._task_spec.dirty_tables) |
| self._initial_issue_count = max(1, len(self._grade.validation_issues)) |
| initial_tables = clone_tables(self._task_spec.dirty_tables) |
| initial_downstream_health = self._compute_downstream_health(self._task_spec, initial_tables, self._grade.validation_issues) |
| self._state = DataCleaningState( |
| episode_id=episode_id or str(uuid4()), |
| step_count=0, |
| task_id=self._task_spec.task_id, |
| task_title=self._task_spec.title, |
| difficulty=self._task_spec.difficulty, |
| requested_seed=normalized_seed, |
| max_steps=self._task_spec.max_steps, |
| review_budget_total=self._task_spec.review_budget, |
| review_budget_remaining=self._task_spec.review_budget, |
| submitted=False, |
| current_score=self._grade.score, |
| best_score=self._grade.score, |
| outstanding_issue_count=len(self._grade.validation_issues), |
| downstream_health=initial_downstream_health, |
| last_dry_run=None, |
| tables=initial_tables, |
| applied_operation_ids=[], |
| inspected_tables=[self._focus_table_name], |
| inspected_operations=[], |
| requested_review_ids=[], |
| pending_reviews=[], |
| resolved_reviews=[], |
| dry_run_targets=[], |
| recent_history=[f"reset -> loaded task {self._task_spec.task_id} ({self._task_spec.difficulty}) seed={normalized_seed}"], |
| ) |
| return self._build_observation( |
| reward_breakdown=RewardBreakdown(total=0.0), |
| reward=0.0, |
| done=False, |
| last_action_status=f"Environment reset to task {self._task_spec.task_id}.", |
| last_action_error=None, |
| ) |
|
|
| def step( |
| self, |
| action: DataCleaningAction, |
| timeout_s: float | None = None, |
| **kwargs: object, |
| ) -> DataCleaningObservation: |
| del timeout_s, kwargs |
| if self._done: |
| penalty = RewardBreakdown(invalid_action_penalty=-0.25, total=-0.25) |
| return self._build_observation( |
| reward_breakdown=penalty, |
| reward=penalty.total, |
| done=True, |
| last_action_status="Episode already finished. Call reset() to start a new task.", |
| last_action_error="Episode already finished. Call reset() to start a new task.", |
| ) |
|
|
| self._state.step_count += 1 |
| previous_score = self._state.current_score |
| previous_issue_count = self._state.outstanding_issue_count |
| previous_downstream_score = self._state.downstream_health.overall_health_score |
|
|
| invalid_action_penalty = 0.0 |
| noop_penalty = 0.0 |
| insight_bonus = 0.0 |
| review_bonus = 0.0 |
| review_cost_penalty = 0.0 |
| action_cost_penalty = 0.0 |
| submit_bonus = 0.0 |
| status_message = "" |
| action_error: str | None = None |
| released_reviews = self._release_ready_reviews() |
| if released_reviews: |
| review_bonus = round(0.04 * len(released_reviews), 4) |
|
|
| if action.action_type == "inspect_table": |
| table_name = normalize_whitespace(action.table_name or "") |
| if table_name not in self._state.tables: |
| invalid_action_penalty = -0.25 |
| status_message = f"Unknown table '{table_name}'." |
| action_error = status_message |
| else: |
| self._focus_table_name = table_name |
| if table_name not in self._state.inspected_tables: |
| self._state.inspected_tables.append(table_name) |
| insight_bonus = 0.01 |
| status_message = f"Inspected table '{table_name}'." |
| else: |
| noop_penalty = -0.02 |
| status_message = f"Table '{table_name}' was already inspected." |
| elif action.action_type == "inspect_operation": |
| operation_id = normalize_whitespace(action.operation_id or "") |
| if operation_id not in self._task_spec.operations: |
| invalid_action_penalty = -0.25 |
| status_message = f"Unknown operation '{operation_id}'." |
| action_error = status_message |
| else: |
| self._focus_operation_detail = self._build_operation_detail(self._task_spec, operation_id, self._state.tables, None) |
| if operation_id not in self._state.inspected_operations: |
| self._state.inspected_operations.append(operation_id) |
| insight_bonus = 0.01 |
| status_message = f"Inspected operation '{operation_id}'." |
| else: |
| noop_penalty = -0.02 |
| status_message = f"Operation '{operation_id}' was already inspected." |
| elif action.action_type == "apply_operation": |
| operation_id = normalize_whitespace(action.operation_id or "") |
| if operation_id not in self._task_spec.operations: |
| invalid_action_penalty = -0.25 |
| status_message = f"Unknown operation '{operation_id}'." |
| action_error = status_message |
| elif operation_id in self._state.applied_operation_ids: |
| noop_penalty = -0.12 |
| self._focus_operation_detail = self._build_operation_detail(self._task_spec, operation_id, self._state.tables, self._state.tables) |
| status_message = f"Operation '{operation_id}' was already applied." |
| else: |
| before_tables = clone_tables(self._state.tables) |
| after_tables = apply_operation_to_tables(self._task_spec, before_tables, operation_id) |
| self._focus_operation_detail = self._build_operation_detail(self._task_spec, operation_id, before_tables, after_tables) |
| if after_tables == before_tables: |
| noop_penalty = -0.08 |
| status_message = f"Operation '{operation_id}' produced no table changes." |
| else: |
| self._state.tables = clone_tables(after_tables) |
| self._state.applied_operation_ids.append(operation_id) |
| affected_tables = ", ".join(self._task_spec.operations[operation_id].tables_affected) |
| if self._task_spec.operations[operation_id].tables_affected: |
| self._focus_table_name = self._task_spec.operations[operation_id].tables_affected[0] |
| status_message = f"Applied '{operation_id}' to {affected_tables or 'current tables'}." |
| elif action.action_type == "request_review": |
| entity_type = normalize_whitespace(action.entity_type or "").lower() |
| entity_id = normalize_whitespace(action.entity_id or "") |
| reason_code = normalize_whitespace(action.reason_code or "") |
| review_case = self._find_review_case(entity_type, entity_id, reason_code) |
| if not entity_type or not entity_id or not reason_code: |
| invalid_action_penalty = -0.25 |
| status_message = "request_review requires entity_type, entity_id, and reason_code." |
| action_error = status_message |
| elif review_case is None: |
| invalid_action_penalty = -0.2 |
| status_message = f"No deterministic review case exists for {entity_type}:{entity_id} ({reason_code})." |
| action_error = status_message |
| elif review_case.review_id in self._state.requested_review_ids: |
| noop_penalty = -0.05 |
| status_message = f"Review '{review_case.review_id}' was already requested." |
| elif self._state.review_budget_remaining <= 0: |
| invalid_action_penalty = -0.18 |
| status_message = "No review budget remaining for this episode." |
| action_error = status_message |
| else: |
| self._state.review_budget_remaining -= 1 |
| self._state.requested_review_ids.append(review_case.review_id) |
| self._state.pending_reviews.append( |
| PendingReview( |
| review_id=review_case.review_id, |
| entity_type=review_case.entity_type, |
| entity_id=review_case.entity_id, |
| reason_code=review_case.reason_code, |
| title=review_case.title, |
| requested_at_step=self._state.step_count, |
| ready_at_step=self._state.step_count + 1, |
| ) |
| ) |
| review_cost_penalty = -0.02 |
| status_message = ( |
| f"Queued review '{review_case.review_id}' for {review_case.entity_type} {review_case.entity_id}; " |
| "response will be available on the next step." |
| ) |
| elif action.action_type == "run_sync_dry_run": |
| target_system = action.target_system |
| if target_system is None: |
| invalid_action_penalty = -0.2 |
| status_message = "run_sync_dry_run requires target_system." |
| action_error = status_message |
| elif target_system not in self._task_spec.sync_targets: |
| invalid_action_penalty = -0.2 |
| status_message = f"Task '{self._task_spec.task_id}' does not support dry-run target '{target_system}'." |
| action_error = status_message |
| else: |
| self._state.last_dry_run = self._build_dry_run_report(target_system) |
| if target_system not in self._state.dry_run_targets: |
| self._state.dry_run_targets.append(target_system) |
| insight_bonus = max(insight_bonus, 0.01) |
| else: |
| noop_penalty = min(noop_penalty, -0.01) |
| status_message = self._state.last_dry_run.summary |
| elif action.action_type == "submit": |
| self._state.submitted = True |
| self._done = True |
| status_message = "Submitted cleaned tables for grading." |
|
|
| action_cost_penalty = -self._estimate_action_cost(action) |
|
|
| self._grade = grade_tables(self._task_spec, self._state.tables) |
| self._state.current_score = self._grade.score |
| self._state.best_score = max(self._state.best_score, self._grade.score) |
| self._state.outstanding_issue_count = len(self._grade.validation_issues) |
| self._state.downstream_health = self._compute_downstream_health(self._task_spec, self._state.tables, self._grade.validation_issues) |
|
|
| quality_delta = round(self._state.current_score - previous_score, 4) |
| issue_delta = round((previous_issue_count - self._state.outstanding_issue_count) / self._initial_issue_count, 4) |
| downstream_health_delta = round(self._state.downstream_health.overall_health_score - previous_downstream_score, 4) |
| efficiency_penalty = -0.01 |
|
|
| if action.action_type == "submit": |
| submission_health = round(0.65 * self._state.current_score + 0.35 * self._state.downstream_health.overall_health_score, 4) |
| submit_bonus = round(0.4 * submission_health, 4) if submission_health >= 0.82 else round(-0.2 * (1.0 - submission_health), 4) |
|
|
| if self._state.step_count >= self._state.max_steps and not self._done: |
| self._done = True |
| self._state.submitted = False |
| status_message = f"{status_message} Step budget exhausted; episode truncated.".strip() |
|
|
| if released_reviews: |
| release_note = ", ".join(review.review_id for review in released_reviews) |
| status_message = f"{status_message} Review response available: {release_note}.".strip() |
|
|
| reward_total = round( |
| 1.0 * quality_delta |
| + 0.35 * issue_delta |
| + 0.55 * downstream_health_delta |
| + insight_bonus |
| + review_bonus |
| + efficiency_penalty |
| + invalid_action_penalty |
| + noop_penalty |
| + review_cost_penalty |
| + action_cost_penalty |
| + submit_bonus, |
| 4, |
| ) |
| reward_breakdown = RewardBreakdown( |
| quality_delta=quality_delta, |
| issue_delta=issue_delta, |
| downstream_health_delta=downstream_health_delta, |
| insight_bonus=insight_bonus, |
| review_bonus=review_bonus, |
| efficiency_penalty=efficiency_penalty, |
| invalid_action_penalty=invalid_action_penalty, |
| noop_penalty=noop_penalty, |
| review_cost_penalty=review_cost_penalty, |
| action_cost_penalty=action_cost_penalty, |
| submit_bonus=submit_bonus, |
| total=reward_total, |
| ) |
|
|
| action_descriptor = action.action_type |
| if action.operation_id: |
| action_descriptor += f"[{action.operation_id}]" |
| if action.table_name: |
| action_descriptor += f"[{action.table_name}]" |
| if action.entity_id: |
| action_descriptor += f"[{action.entity_id}]" |
| if action.target_system: |
| action_descriptor += f"[{action.target_system}]" |
| self._state.recent_history.append(f"step {self._state.step_count}: {action_descriptor} -> score={self._state.current_score:.4f}") |
| self._state.recent_history = self._state.recent_history[-10:] |
|
|
| return self._build_observation( |
| reward_breakdown=reward_breakdown, |
| reward=reward_total, |
| done=self._done, |
| last_action_status=status_message or "Action processed.", |
| last_action_error=action_error, |
| ) |
|
|
| @property |
| def state(self) -> DataCleaningState: |
| return self._state |
|
|
| def get_metadata(self) -> EnvironmentMetadata: |
| return EnvironmentMetadata( |
| name="CleanOpsEnvironment", |
| description="A realistic OpenEnv benchmark where an agent cleans operational customer, order, subscription, and payment tables using a curated data-cleaning toolkit.", |
| version="0.1.0", |
| author="OpenEnv CleanOps", |
| ) |
|
|
| def _build_observation( |
| self, |
| *, |
| reward_breakdown: RewardBreakdown, |
| reward: float, |
| done: bool, |
| last_action_status: str, |
| last_action_error: str | None, |
| ) -> DataCleaningObservation: |
| summaries = [build_table_summary(self._task_spec, table_name, self._state.tables) for table_name in self._task_spec.dirty_tables] |
| focus_table = self._build_table_view(self._task_spec, self._focus_table_name) |
| available_operations = [ |
| OperationSummary( |
| operation_id=operation.operation_id, |
| title=operation.title, |
| category=operation.category, |
| risk=operation.risk, |
| tables_affected=list(operation.tables_affected), |
| description=operation.description, |
| already_applied=operation.operation_id in self._state.applied_operation_ids, |
| ) |
| for operation in sorted(self._task_spec.operations.values(), key=lambda op: op.operation_id) |
| ] |
| available_review_targets = [ |
| ReviewTarget( |
| review_id=review_case.review_id, |
| entity_type=review_case.entity_type, |
| entity_id=review_case.entity_id, |
| reason_code=review_case.reason_code, |
| title=review_case.title, |
| detail=review_case.detail, |
| recommended_operation_ids=list(review_case.recommended_operation_ids), |
| ) |
| for review_case in sorted(self._task_spec.review_cases.values(), key=lambda case: case.review_id) |
| ] |
| return DataCleaningObservation( |
| task_id=self._task_spec.task_id, |
| task_title=self._task_spec.title, |
| difficulty=self._task_spec.difficulty, |
| requested_seed=self._state.requested_seed, |
| objective=self._task_spec.objective, |
| dataset_context=self._task_spec.dataset_context, |
| quality_score=self._state.current_score, |
| best_score=self._state.best_score, |
| remaining_steps=max(0, self._state.max_steps - self._state.step_count), |
| review_budget_remaining=self._state.review_budget_remaining, |
| supported_sync_targets=list(self._task_spec.sync_targets), |
| downstream_health=self._state.downstream_health, |
| risk_cards=self._build_risk_cards(), |
| last_dry_run=self._state.last_dry_run, |
| action_costs=self._build_action_cost_entries(), |
| table_summaries=summaries, |
| focus_table=focus_table, |
| available_operations=available_operations, |
| available_review_targets=available_review_targets, |
| pending_reviews=list(self._state.pending_reviews), |
| resolved_reviews=list(self._state.resolved_reviews), |
| focus_operation=self._focus_operation_detail, |
| validation_issues=self._grade.validation_issues, |
| issue_cards=list(self._task_spec.issue_cards), |
| recent_history=list(self._state.recent_history), |
| grader=self._grade.breakdown, |
| reward_breakdown=reward_breakdown, |
| last_action_status=last_action_status, |
| last_action_error=last_action_error, |
| reward=reward, |
| done=done, |
| metadata={ |
| "episode_id": self._state.episode_id, |
| "requested_seed": self._state.requested_seed, |
| "applied_operation_ids": list(self._state.applied_operation_ids), |
| "review_budget_remaining": self._state.review_budget_remaining, |
| "requested_review_ids": list(self._state.requested_review_ids), |
| "dry_run_targets": list(self._state.dry_run_targets), |
| "submitted": self._state.submitted, |
| }, |
| ) |
|
|
| def _build_table_view(self, task_spec: TaskSpec, table_name: str) -> TableView: |
| primary_key = task_spec.primary_keys[table_name] |
| rows = self._preview_rows(task_spec, table_name, self._state.tables.get(table_name, [])) |
| columns = sorted({column_name for row in rows for column_name in row}) |
| return TableView(name=table_name, primary_key=primary_key, columns=columns, rows=rows) |
|
|
| def _choose_initial_focus_table(self, task_spec: TaskSpec, seed: int | None) -> str: |
| table_names = sorted(task_spec.dirty_tables) |
| if not table_names: |
| return first_table_name(task_spec) |
| if seed is None: |
| return table_names[0] |
| return table_names[seed % len(table_names)] |
|
|
| def _preview_rows( |
| self, |
| task_spec: TaskSpec, |
| table_name: str, |
| rows: list[dict[str, str]], |
| ) -> list[dict[str, str]]: |
| primary_key = task_spec.primary_keys[table_name] |
| ordered_rows = sorted_rows(rows, primary_key) |
| seed = self._state.requested_seed |
| if seed is None or len(ordered_rows) <= 1: |
| return ordered_rows |
| shuffled_rows = copy.deepcopy(ordered_rows) |
| random.Random(seed + sum(ord(char) for char in table_name)).shuffle(shuffled_rows) |
| return shuffled_rows |
|
|
| def _find_review_case(self, entity_type: str, entity_id: str, reason_code: str) -> ReviewCaseSpec | None: |
| for review_case in self._task_spec.review_cases.values(): |
| if ( |
| review_case.entity_type == entity_type |
| and review_case.entity_id == entity_id |
| and review_case.reason_code == reason_code |
| ): |
| return review_case |
| return None |
|
|
| def _release_ready_reviews(self) -> list[ReviewResolution]: |
| if not self._state.pending_reviews: |
| return [] |
|
|
| still_pending: list[PendingReview] = [] |
| released: list[ReviewResolution] = [] |
| for pending_review in self._state.pending_reviews: |
| if pending_review.ready_at_step > self._state.step_count: |
| still_pending.append(pending_review) |
| continue |
| review_case = self._task_spec.review_cases[pending_review.review_id] |
| released_review = ReviewResolution( |
| review_id=review_case.review_id, |
| entity_type=review_case.entity_type, |
| entity_id=review_case.entity_id, |
| reason_code=review_case.reason_code, |
| title=review_case.title, |
| resolution=review_case.resolution, |
| response_summary=review_case.response_summary, |
| evidence_summary=review_case.evidence_summary, |
| recommended_operation_ids=list(review_case.recommended_operation_ids), |
| ) |
| self._state.resolved_reviews.append(released_review) |
| released.append(released_review) |
| self._state.pending_reviews = still_pending |
| return released |
|
|
| def _estimate_action_cost(self, action: DataCleaningAction) -> float: |
| if action.action_type == "apply_operation": |
| operation = self._task_spec.operations.get(normalize_whitespace(action.operation_id or "")) |
| if operation is None: |
| return ACTION_COSTS["apply_operation:safe"] |
| if operation.risk == "review": |
| return ACTION_COSTS["apply_operation:review"] |
| if operation.risk == "destructive": |
| return ACTION_COSTS["apply_operation:destructive"] |
| return ACTION_COSTS["apply_operation:safe"] |
| return ACTION_COSTS.get(action.action_type, 0.01) |
|
|
| def _build_action_cost_entries(self) -> list[ActionCostEntry]: |
| return [ |
| ActionCostEntry(action_key=action_key, estimated_cost=estimated_cost, description=ACTION_COST_DESCRIPTIONS[action_key]) |
| for action_key, estimated_cost in ACTION_COSTS.items() |
| ] |
|
|
| @staticmethod |
| def _open_metric(value: float) -> float: |
| return round(min(0.99, max(0.01, value)), 4) |
|
|
| def _compute_downstream_health( |
| self, |
| task_spec: TaskSpec, |
| tables: dict[str, list[dict[str, str]]], |
| validation_issues: list, |
| ) -> DownstreamHealth: |
| customers = tables.get("customers", []) |
| orders = tables.get("orders", []) |
| subscriptions = tables.get("subscriptions", []) |
| payments = tables.get("payments", []) |
|
|
| crm_rows = max(1, len(customers) + len(subscriptions)) |
| billing_rows = max(1, len(orders) + len(subscriptions) + len(payments)) |
| payment_rows = max(1, len(orders) + len(payments)) |
|
|
| crm_issue_weight = sum(max(1, len(issue.row_ids)) for issue in validation_issues if issue.table_name in {"customers", "subscriptions"}) |
| billing_issue_weight = sum( |
| max(1, len(issue.row_ids)) |
| for issue in validation_issues |
| if issue.table_name in {"orders", "payments", "subscriptions"} |
| and (issue.code.startswith("foreign_key:") or issue.code.startswith("required:") or issue.code.startswith("unique:")) |
| ) |
| payment_issue_weight = sum( |
| max(1, len(issue.row_ids)) |
| for issue in validation_issues |
| if issue.table_name in {"orders", "payments"} |
| ) |
|
|
| customer_duplicate_groups = count_duplicate_groups(task_spec, "customers", customers) if "customers" in task_spec.duplicate_identity_columns else 0 |
| customer_rows = max(1, len(customers)) |
| payment_duplicate_groups = count_duplicate_groups(task_spec, "payments", payments) if "payments" in task_spec.duplicate_identity_columns else 0 |
|
|
| crm_sync_success_rate = self._open_metric(1.0 - (crm_issue_weight / max(2, crm_rows * 2))) |
| if not orders and not payments: |
| billing_link_integrity = 0.99 |
| revenue_reporting_risk = 0.01 |
| else: |
| billing_link_integrity = self._open_metric(1.0 - (billing_issue_weight / max(2, billing_rows * 2))) |
| revenue_reporting_risk = self._open_metric(min(0.99, (payment_issue_weight / max(2, payment_rows * 2)) + (payment_duplicate_groups / max(1, payment_rows)))) |
|
|
| duplicate_contact_risk = self._open_metric(min(0.99, (customer_duplicate_groups / customer_rows) + 0.06 * sum(1 for issue in validation_issues if issue.code.startswith("unique:customers")))) |
| overall_health_score = self._open_metric( |
| ( |
| crm_sync_success_rate |
| + billing_link_integrity |
| + (1.0 - duplicate_contact_risk) |
| + (1.0 - revenue_reporting_risk) |
| ) |
| / 4.0 |
| ) |
|
|
| return DownstreamHealth( |
| crm_sync_success_rate=crm_sync_success_rate, |
| billing_link_integrity=billing_link_integrity, |
| duplicate_contact_risk=duplicate_contact_risk, |
| revenue_reporting_risk=revenue_reporting_risk, |
| overall_health_score=overall_health_score, |
| ) |
|
|
| def _build_risk_cards(self) -> list[RiskCard]: |
| health = self._state.downstream_health |
| cards = [ |
| RiskCard( |
| title="CRM import risk", |
| detail="Customer and subscription issues can block CRM migration syncs.", |
| severity="high" if health.crm_sync_success_rate < 0.8 else "medium" if health.crm_sync_success_rate < 0.92 else "low", |
| metric_name="crm_sync_success_rate", |
| current_value=health.crm_sync_success_rate, |
| recommended_action_ids=[op_id for op_id in self._recommended_operation_ids_for_tables({"customers", "subscriptions"})], |
| ), |
| RiskCard( |
| title="Billing linkage risk", |
| detail="Broken foreign keys or missing IDs can mislink orders, subscriptions, and payments.", |
| severity="high" if health.billing_link_integrity < 0.8 else "medium" if health.billing_link_integrity < 0.92 else "low", |
| metric_name="billing_link_integrity", |
| current_value=health.billing_link_integrity, |
| recommended_action_ids=[op_id for op_id in self._recommended_operation_ids_for_tables({"orders", "subscriptions", "payments"})], |
| ), |
| RiskCard( |
| title="Duplicate contact risk", |
| detail="Remaining duplicate customer identities can create bad merges downstream.", |
| severity="high" if health.duplicate_contact_risk > 0.3 else "medium" if health.duplicate_contact_risk > 0.12 else "low", |
| metric_name="duplicate_contact_risk", |
| current_value=health.duplicate_contact_risk, |
| recommended_action_ids=[op_id for op_id in self._recommended_operation_ids_for_keyword("merge")], |
| ), |
| RiskCard( |
| title="Revenue reporting risk", |
| detail="Duplicate or mislinked payment and order facts can distort downstream reporting.", |
| severity="high" if health.revenue_reporting_risk > 0.3 else "medium" if health.revenue_reporting_risk > 0.12 else "low", |
| metric_name="revenue_reporting_risk", |
| current_value=health.revenue_reporting_risk, |
| recommended_action_ids=[op_id for op_id in self._recommended_operation_ids_for_tables({"orders", "payments"})], |
| ), |
| ] |
| return cards |
|
|
| def _recommended_operation_ids_for_tables(self, table_names: set[str]) -> list[str]: |
| return [ |
| operation.operation_id |
| for operation in sorted(self._task_spec.operations.values(), key=lambda op: op.operation_id) |
| if set(operation.tables_affected) & table_names |
| ][:4] |
|
|
| def _recommended_operation_ids_for_keyword(self, keyword: str) -> list[str]: |
| lowered = keyword.lower() |
| return [ |
| operation.operation_id |
| for operation in sorted(self._task_spec.operations.values(), key=lambda op: op.operation_id) |
| if lowered in operation.operation_id.lower() or lowered in operation.title.lower() |
| ][:4] |
|
|
| def _build_dry_run_report(self, target_system: str) -> DryRunReport: |
| findings: list[DryRunFinding] = [] |
| for issue in self._grade.validation_issues: |
| if target_system == "crm" and issue.table_name not in {"customers", "subscriptions"}: |
| continue |
| if target_system == "billing" and issue.table_name not in {"orders", "subscriptions", "payments"}: |
| continue |
| findings.append( |
| DryRunFinding( |
| code=issue.code, |
| severity=issue.severity, |
| table_name=issue.table_name, |
| row_ids=list(issue.row_ids), |
| message=issue.message, |
| ) |
| ) |
|
|
| health = self._state.downstream_health |
| success_rate = health.crm_sync_success_rate if target_system == "crm" else health.billing_link_integrity |
|
|
| if target_system == "crm" and health.duplicate_contact_risk > 0.12: |
| findings.append( |
| DryRunFinding( |
| code="risk:duplicate_contacts", |
| severity="medium" if health.duplicate_contact_risk <= 0.3 else "high", |
| table_name="customers", |
| message="CRM dry run predicts duplicate-contact collisions after import.", |
| ) |
| ) |
| if target_system == "billing" and health.revenue_reporting_risk > 0.12: |
| findings.append( |
| DryRunFinding( |
| code="risk:revenue_reporting", |
| severity="medium" if health.revenue_reporting_risk <= 0.3 else "high", |
| table_name="payments" if "payments" in self._state.tables else "orders", |
| message="Billing dry run predicts mislinked or duplicated revenue facts.", |
| ) |
| ) |
|
|
| summary = ( |
| f"Dry run for {target_system.upper()} found {len(findings)} blocking or risky findings; " |
| f"estimated success rate is {success_rate:.2f}." |
| ) |
| return DryRunReport( |
| target_system=target_system, |
| success_rate=success_rate, |
| finding_count=len(findings), |
| findings=findings, |
| summary=summary, |
| generated_at_step=self._state.step_count, |
| ) |
|
|
| def _build_operation_detail( |
| self, |
| task_spec: TaskSpec, |
| operation_id: str, |
| before_tables: dict[str, list[dict[str, str]]], |
| after_tables: dict[str, list[dict[str, str]]] | None, |
| ) -> OperationDetail: |
| operation = task_spec.operations[operation_id] |
| simulated_after = after_tables |
| if simulated_after is None: |
| simulated_after = apply_operation_to_tables(task_spec, before_tables, operation_id) |
|
|
| preview: list[RowChange] = [] |
| for table_name in operation.tables_affected: |
| primary_key = task_spec.primary_keys[table_name] |
| before_rows = {normalize_whitespace(row.get(primary_key, "")): dict(row) for row in before_tables.get(table_name, [])} |
| after_rows = {normalize_whitespace(row.get(primary_key, "")): dict(row) for row in simulated_after.get(table_name, [])} |
| changed_keys = sorted(set(before_rows) | set(after_rows)) |
| for row_key in changed_keys: |
| if before_rows.get(row_key) == after_rows.get(row_key): |
| continue |
| preview.append(RowChange(primary_key_value=row_key, before=before_rows.get(row_key), after=after_rows.get(row_key))) |
| if len(preview) >= 12: |
| break |
| if len(preview) >= 12: |
| break |
|
|
| return OperationDetail( |
| operation_id=operation.operation_id, |
| title=operation.title, |
| category=operation.category, |
| risk=operation.risk, |
| tables_affected=list(operation.tables_affected), |
| description=operation.description, |
| already_applied=operation.operation_id in self._state.applied_operation_ids, |
| why_it_matters=operation.why_it_matters, |
| change_preview=preview, |
| ) |
|
|
