Coverage for backend/django/Economics/results/services/lifecycle/runs.py: 100%

1"""Result-run reuse, classification, and stale-state transitions. 

2 

3This module owns status decisions for persisted result runs. It compares stored 

4fingerprints with current source fingerprints, annotates reuse/stale events, and 

5must not calculate new economics metrics or materialize result lines. 

6""" 

7 

8from __future__ import annotations 

9 

10from django.db import transaction 

11from django.utils import timezone 

12 

13from core.validation import flowsheet_context, get_current_flowsheet 

14from Economics.results.models import EconomicsResultRun 

15 

16from Economics.studies.models import EconomicsStudy 

17from Economics.studies.services.baseline_access import stable_flowsheet_id 

18 

19from Economics.shared.choices import ResultDependencyType, ResultRunStatus 

20from Economics.results.services.lifecycle.fingerprints import ( 

21 DependencyFingerprint, 

22 build_dependency_fingerprints, 

23 _fingerprint_map, 

24 _stored_dependency_map, 

25 promote_legacy_schedule_source_version, 

26 schedule_source_version, 

27) 

28 

29 

30class ResultRunClassification: 

31 """Stable result-run classification labels exposed through serializers.""" 

32 

33 CURRENT = "current" 

34 STALE = "stale" 

35 REUSABLE = "reusable" 

36 

37 

38 

39def classify_result_run(result_run: EconomicsResultRun) -> str: 

40 """Classify a persisted run against the study's current dependency state.""" 

41 context = get_current_flowsheet() or {} 

42 user = context.get("user") 

43 active_flowsheet_id = context.get("flowsheet") 

44 result_flowsheet_id = stable_flowsheet_id(result_run.study) 

45 if user is not None and active_flowsheet_id != result_flowsheet_id: 

46 # Result-run classification must compare dependencies from the run's 

47 # owning flowsheet. This matters when a target study uses a readable 

48 # cross-flowsheet baseline: the target request context would otherwise 

49 # hide the baseline study's dependencies and make a current run look 

50 # stale. Re-entering the run flowsheet keeps the normal access-control 

51 # managers in play; it does not broaden what the user can read. 

52 with flowsheet_context(result_flowsheet_id, user): 

53 return classify_result_run(result_run) 

54 dependencies = list(result_run.dependencies.order_by("dependency_type", "dependency_key")) 

55 schedule_dependency = next( 

56 ( 

57 dependency 

58 for dependency in dependencies 

59 if dependency.dependency_type == ResultDependencyType.SCHEDULE 

60 ), 

61 None, 

62 ) 

63 current_schedule_version = schedule_source_version(result_run.study) 

64 persisted_schedule_fingerprint = "" 

65 if ( 

66 schedule_dependency is not None 

67 and current_schedule_version 

68 and schedule_dependency.source_version == current_schedule_version 

69 ): 

70 persisted_schedule_fingerprint = schedule_dependency.fingerprint_value 

71 current_schedule_version = promote_legacy_schedule_source_version( 

72 result_run.study, 

73 verified_source_version=current_schedule_version, 

74 ) 

75 current_map = _fingerprint_map( 

76 build_dependency_fingerprints( 

77 result_run.study, 

78 persisted_schedule_fingerprint=persisted_schedule_fingerprint, 

79 schedule_version=current_schedule_version, 

80 ) 

81 ) 

82 run_map = _stored_dependency_map(result_run, dependencies=dependencies) 

83 if run_map == current_map: 

84 if result_run.status == ResultRunStatus.CURRENT: 

85 return ResultRunClassification.CURRENT 

86 return ResultRunClassification.REUSABLE 

87 return ResultRunClassification.STALE 

88 

89 

90def mark_result_runs_stale_for_study( 

91 *, 

92 study: EconomicsStudy, 

93 reason: str, 

94 requires_solve: bool = False, 

95) -> int: 

96 """Mark current runs stale with diagnostic context, without calculating new results.""" 

97 with transaction.atomic(): 

98 runs = list( 

99 EconomicsResultRun.objects.select_for_update().filter( 

100 flowsheet_state=study.flowsheet_state, 

101 study=study, 

102 status=ResultRunStatus.CURRENT, 

103 ) 

104 ) 

105 for run in runs: 

106 _mark_run_stale(run=run, reason=reason, requires_solve=requires_solve) 

107 return len(runs) 

108 

109 

110def _matching_result_run( 

111 *, 

112 study: EconomicsStudy, 

113 fingerprints: list[DependencyFingerprint], 

114) -> EconomicsResultRun | None: 

115 """Return the newest current/stale run whose stored dependencies still match.""" 

116 current_map = _fingerprint_map(fingerprints) 

117 for status in (ResultRunStatus.CURRENT, ResultRunStatus.STALE): 

118 for result_run in EconomicsResultRun.objects.filter( 

119 flowsheet_state=study.flowsheet_state, 

120 study=study, 

121 status=status, 

122 ).order_by("-created_at"): 

123 if _stored_dependency_map(result_run) == current_map: 

124 return result_run 

125 return None 

126 

127 

128def _mark_nonmatching_current_runs_stale( 

129 *, 

130 study: EconomicsStudy, 

131 fingerprints: list[DependencyFingerprint], 

132 reason: str, 

133) -> None: 

134 """Mark only current runs stale when their dependency map no longer matches.""" 

135 current_map = _fingerprint_map(fingerprints) 

136 for result_run in EconomicsResultRun.objects.select_for_update().filter( 

137 flowsheet_state=study.flowsheet_state, 

138 study=study, 

139 status=ResultRunStatus.CURRENT, 

140 ): 

141 if _stored_dependency_map(result_run) != current_map: 

142 _mark_run_stale(run=result_run, reason=reason, requires_solve=False) 

143 

144 

145def _mark_run_stale(*, run: EconomicsResultRun, reason: str, requires_solve: bool) -> None: 

146 """Append a stale event without discarding existing warning diagnostics.""" 

147 payload = dict(run.warning_payload or {}) 

148 stale_events = list(payload.get("stale_events", [])) 

149 stale_events.append( 

150 { 

151 "reason": reason, 

152 "requires_solve": requires_solve, 

153 "marked_at": timezone.now().isoformat(), 

154 "previous_status": run.status, 

155 } 

156 ) 

157 payload["stale_events"] = stale_events 

158 payload["latest_stale_reason"] = reason 

159 payload["requires_solve"] = requires_solve 

160 run.status = ResultRunStatus.STALE 

161 run.warning_payload = payload 

162 run.save(update_fields=["status", "warning_payload"]) 

163 

164 

165def _annotate_reused_run(*, result_run: EconomicsResultRun, reason: str, duration_ms: int) -> None: 

166 """Record reuse diagnostics after a matching run is promoted or refreshed.""" 

167 payload = dict(result_run.warning_payload or {}) 

168 reuse_events = list(payload.get("reuse_events", [])) 

169 reuse_events.append( 

170 { 

171 "reason": reason, 

172 "reused_at": timezone.now().isoformat(), 

173 "duration_ms": duration_ms, 

174 } 

175 ) 

176 payload["reuse_events"] = reuse_events 

177 result_run.warning_payload = payload 

178 result_run.save(update_fields=["warning_payload"])