Coverage for scripts / live_release_validation / ownership / stacks.py: 100.00%
265 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +0000
1"""Durable CloudFormation stack ownership and change-set authority.
3A stack is normally owned through *prepared-change-set* authority: the harness
4prepared the change set itself and checkpointed its identity before
5CloudFormation executed anything. The upgrade harness cannot do that, because
6the ``gco`` it validates deploys the stacks from a separate process. For that
7harness alone (``RunSettings.allows_run_tag_adoption``) a stack can instead be
8*adopted by run tag*: it carries this run's exact ``GcoLiveValidationRun`` tag,
9its name is a checkpointed target, and CloudFormation reports it created after
10the adopting phase began. Every destructive boundary still re-checks the exact
11stack ID and tag, as it does for prepared stacks.
12"""
14from __future__ import annotations
16import copy
17import json
18from collections.abc import Collection
19from datetime import UTC, datetime, timedelta
20from typing import Any
22from ..constants import (
23 _RUN_STACK_TAG,
24)
25from ..inventory import (
26 collect_project_stacks,
27 describe_stack,
28)
29from ..models import RunContext, utc_now
31# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit
32# Generated at (UTC): 2026-09-29T11:35:23Z
33# Generated from Git commit: a081fbf678ca3f3fa3f7edb74687de923efe6657
34# Flowchart(s) generated from this file:
35# * ``_adopt_run_tagged_stacks`` -> ``diagrams/code_diagrams/scripts/live_release_validation/ownership/stacks._adopt_run_tagged_stacks.html``
36# (PNG: ``diagrams/code_diagrams/scripts/live_release_validation/ownership/stacks._adopt_run_tagged_stacks.png``)
37# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``.
38# <pyflowchart-code-diagram> END
41#: ``authority`` of a stack record adopted by run tag rather than prepared here.
42_RUN_TAG_ADOPTION_AUTHORITY = "run-tag-adoption"
44#: The adoption window starts at a server ``Date`` header, which has one-second
45#: resolution; CloudFormation reports creation times to the millisecond.
46_ADOPTION_CLOCK_RESOLUTION = timedelta(seconds=1)
49def _run_tag_adoption_allowed(ctx: RunContext) -> bool:
50 """Only settings that opt in (the upgrade harness) may adopt stacks by run tag."""
51 return getattr(ctx.settings, "allows_run_tag_adoption", False) is True
54def _owned_stacks(ctx: RunContext) -> dict[str, dict[str, dict[str, Any]]]:
55 """Return region-qualified stack ownership records for this schema."""
56 owned = ctx.checkpoint.state.setdefault("owned_stacks", {})
57 if not isinstance(owned, dict):
58 raise RuntimeError("Checkpoint owned_stacks must be an object")
59 for region, records in owned.items():
60 if not isinstance(records, dict):
61 raise RuntimeError(f"Checkpoint stack ownership for {region} is malformed")
62 for stack_name, record in records.items():
63 if not isinstance(record, dict):
64 raise RuntimeError(
65 f"Checkpoint stack ownership for {region}:{stack_name} is malformed"
66 )
67 return owned
70def _owned_stack_record(
71 ctx: RunContext,
72 region: str,
73 stack_name: str,
74) -> dict[str, Any] | None:
75 return _owned_stacks(ctx).get(region, {}).get(stack_name)
78def _require_prepared_stack_authority(
79 record: dict[str, Any],
80 *,
81 region: str,
82 stack_name: str,
83) -> None:
84 if (
85 record.get("authority") != "prepared-change-set"
86 or not record.get("change_set_id")
87 or record.get("change_set_type") not in {"CREATE", "UPDATE"}
88 ):
89 raise RuntimeError(
90 f"Stack {region}:{stack_name} lacks persisted prepared-change-set authority"
91 )
94def _is_adopted(record: dict[str, Any] | None) -> bool:
95 return record is not None and record.get("authority") == _RUN_TAG_ADOPTION_AUTHORITY
98def _require_stack_authority(
99 ctx: RunContext,
100 record: dict[str, Any],
101 *,
102 region: str,
103 stack_name: str,
104) -> None:
105 """Require prepared-change-set authority, or run-tag adoption where it is allowed."""
106 if not _is_adopted(record):
107 _require_prepared_stack_authority(record, region=region, stack_name=stack_name)
108 return
109 if not _run_tag_adoption_allowed(ctx):
110 raise RuntimeError(
111 f"Stack {region}:{stack_name} carries run-tag adoption authority, which only the "
112 "upgrade validation harness may honor"
113 )
114 adoption = record.get("adoption")
115 if (
116 not isinstance(adoption, dict)
117 or not adoption.get("phase")
118 or not adoption.get("window_started_at")
119 or not adoption.get("creation_time")
120 ):
121 raise RuntimeError(f"Stack {region}:{stack_name} lacks persisted run-tag adoption evidence")
124def _owned_stack_ids(ctx: RunContext, region: str, stack_name: str) -> frozenset[str]:
125 """Every stack ID this run has owned under one name.
127 That is the current generation plus, for an adopted stack, each generation
128 an in-place upgrade replaced. A retained resource checkpointed from a
129 replaced generation (its EKS key, its log groups) stays attributable to
130 the run after that generation is gone.
131 """
132 record = _owned_stack_record(ctx, region, stack_name)
133 if record is None:
134 return frozenset()
135 identities = {str(record.get("stack_id") or "")}
136 generations = record.get("replaced_generations", [])
137 if not isinstance(generations, list) or (generations and not _is_adopted(record)):
138 raise RuntimeError(f"Replaced stack generations for {region}:{stack_name} are malformed")
139 if generations and not _run_tag_adoption_allowed(ctx):
140 raise RuntimeError(
141 f"Stack {region}:{stack_name} records replaced generations, which only the "
142 "upgrade validation harness may honor"
143 )
144 for generation in generations:
145 stack_id = generation.get("stack_id") if isinstance(generation, dict) else None
146 if not isinstance(stack_id, str) or not stack_id:
147 raise RuntimeError(
148 f"Replaced stack generations for {region}:{stack_name} are malformed"
149 )
150 identities.add(stack_id)
151 identities.discard("")
152 return frozenset(identities)
155def _prepared_change_set_authority(
156 ctx: RunContext,
157) -> dict[str, dict[str, dict[str, str]]]:
158 """Return validated per-target preparation history, including legacy checkpoints."""
159 target_regions = ctx.checkpoint.state.get("target_stack_regions")
160 if not isinstance(target_regions, dict):
161 raise RuntimeError("Checkpoint target_stack_regions must be an object")
163 authority: dict[str, dict[str, dict[str, str]]] = {}
164 for stack_name, region_value in target_regions.items():
165 region = str(region_value)
166 record = _owned_stack_record(ctx, region, stack_name)
167 prepared_records: dict[str, dict[str, str]] = {}
168 if record is not None:
169 _require_stack_authority(
170 ctx,
171 record,
172 region=region,
173 stack_name=stack_name,
174 )
175 raw_records = record.get("prepared_change_sets", {})
176 if not isinstance(raw_records, dict):
177 raise RuntimeError(
178 f"Prepared change-set history for {region}:{stack_name} is malformed"
179 )
180 for change_set_id, raw_prepared in raw_records.items():
181 if not isinstance(change_set_id, str) or not isinstance(raw_prepared, dict):
182 raise RuntimeError(
183 f"Prepared change-set history for {region}:{stack_name} is malformed"
184 )
185 prepared = {
186 "change_set_id": str(raw_prepared.get("change_set_id") or ""),
187 "stack_id": str(raw_prepared.get("stack_id") or ""),
188 "change_set_type": str(raw_prepared.get("change_set_type") or ""),
189 }
190 if (
191 prepared["change_set_id"] != change_set_id
192 or prepared["stack_id"] != str(record.get("stack_id") or "")
193 or prepared["change_set_type"] not in {"CREATE", "UPDATE"}
194 ):
195 raise RuntimeError(
196 f"Prepared change-set history for {region}:{stack_name} is inconsistent"
197 )
198 prepared_records[change_set_id] = prepared
200 # Checkpoints written before per-change-set history retained only
201 # the latest preparation. Preserve that exact authority on resume.
202 legacy_change_set_id = str(record.get("change_set_id") or "")
203 if legacy_change_set_id and legacy_change_set_id not in prepared_records:
204 prepared_records[legacy_change_set_id] = {
205 "change_set_id": legacy_change_set_id,
206 "stack_id": str(record.get("stack_id") or ""),
207 "change_set_type": str(record.get("change_set_type") or ""),
208 }
209 authority[stack_name] = prepared_records
210 return authority
213def _record_prepared_stack_identity(
214 ctx: RunContext,
215 stack_name: str,
216 region: str,
217 stack_id: str,
218 change_set_id: str,
219 change_set_type: str,
220) -> None:
221 """Persist causal change-set authority before CloudFormation execution."""
222 if not stack_id or not change_set_id or change_set_type not in {"CREATE", "UPDATE"}:
223 raise RuntimeError(f"Invalid prepared change-set identity for {region}:{stack_name}")
224 with ctx.state_lock:
225 records = _owned_stacks(ctx).setdefault(region, {})
226 previous = records.get(stack_name)
227 core = {"name": stack_name, "region": region, "stack_id": stack_id}
228 if previous is not None:
229 _require_stack_authority(
230 ctx,
231 previous,
232 region=region,
233 stack_name=stack_name,
234 )
235 if any(previous.get(key) != value for key, value in core.items()):
236 raise RuntimeError(
237 f"Prepared stack identity changed for {region}:{stack_name}; refusing adoption"
238 )
239 previous_prepared = (previous or {}).get("prepared_change_sets", {})
240 if not isinstance(previous_prepared, dict):
241 raise RuntimeError(
242 f"Prepared change-set history for {region}:{stack_name} is malformed"
243 )
244 prepared_records = copy.deepcopy(previous_prepared)
245 # Prepared records always carry the latest change set; an adopted record
246 # carries none until the harness prepares one on the stack itself.
247 if previous is not None and previous.get("change_set_id"):
248 legacy_change_set_id = str(previous.get("change_set_id") or "")
249 legacy_record = {
250 "change_set_id": legacy_change_set_id,
251 "stack_id": stack_id,
252 "change_set_type": str(previous.get("change_set_type") or ""),
253 }
254 persisted_legacy = prepared_records.get(legacy_change_set_id)
255 if persisted_legacy is not None and persisted_legacy != legacy_record:
256 raise RuntimeError(
257 f"Prepared change-set history for {region}:{stack_name} is inconsistent"
258 )
259 prepared_records[legacy_change_set_id] = legacy_record
260 prepared_record = {
261 "change_set_id": change_set_id,
262 "stack_id": stack_id,
263 "change_set_type": change_set_type,
264 }
265 existing_prepared = prepared_records.get(change_set_id)
266 if existing_prepared is not None and existing_prepared != prepared_record:
267 raise RuntimeError(
268 f"Prepared change-set identity changed for {region}:{stack_name}; refusing adoption"
269 )
270 prepared_records[change_set_id] = prepared_record
271 records[stack_name] = {
272 **(previous or {}),
273 **core,
274 "run_tag": ctx.settings.run_id,
275 # A change set the harness prepares on an adopted stack extends its
276 # history; the adoption stays the record's basis of ownership.
277 "authority": (
278 _RUN_TAG_ADOPTION_AUTHORITY if _is_adopted(previous) else "prepared-change-set"
279 ),
280 "change_set_id": change_set_id,
281 "change_set_type": change_set_type,
282 "prepared_change_sets": prepared_records,
283 }
284 ctx.persist_callback(ctx.checkpoint)
287def _record_stack_identity(
288 ctx: RunContext,
289 stack_name: str,
290 region: str,
291 stack: dict[str, Any],
292) -> dict[str, Any]:
293 stack_id = str(stack.get("stack_id") or "")
294 run_tag = str((stack.get("tags") or {}).get(_RUN_STACK_TAG) or "")
295 if stack.get("name") != stack_name or not stack_id:
296 raise RuntimeError(f"CloudFormation returned an invalid identity for {region}:{stack_name}")
297 if run_tag != ctx.settings.run_id:
298 raise RuntimeError(
299 f"Stack {region}:{stack_name} is not tagged for run {ctx.settings.run_id!r}"
300 )
302 with ctx.state_lock:
303 records = _owned_stacks(ctx).get(region)
304 if records is None:
305 raise RuntimeError(
306 f"Stack {region}:{stack_name} was observed without prepared-change-set authority"
307 )
308 previous = records.get(stack_name)
309 if previous is None:
310 raise RuntimeError(
311 f"Stack {region}:{stack_name} was observed without prepared-change-set authority"
312 )
313 _require_stack_authority(
314 ctx,
315 previous,
316 region=region,
317 stack_name=stack_name,
318 )
319 core = {
320 "name": stack_name,
321 "region": region,
322 "stack_id": stack_id,
323 "run_tag": run_tag,
324 }
325 if any(previous.get(key) != value for key, value in core.items()):
326 raise RuntimeError(
327 f"Stack identity changed for {region}:{stack_name}; refusing name-based adoption"
328 )
329 candidate = {**previous, **core}
330 records[stack_name] = candidate
331 ctx.persist_callback(ctx.checkpoint)
332 return candidate
335def _reconcile_stack_ownership(ctx: RunContext) -> dict[str, Any]:
336 """Verify every live project stack by ARN and exact run tag."""
337 target_regions = ctx.checkpoint.state.get("target_stack_regions") or {}
338 enabled_regions = ctx.checkpoint.state.get("enabled_regions") or []
339 if not target_regions or not enabled_regions:
340 raise RuntimeError("Checkpoint lacks target stack Regions or enabled Regions")
342 project_stacks = collect_project_stacks(
343 ctx.session,
344 enabled_regions,
345 ctx.config.project_name,
346 )
347 expected_targets = {
348 (str(region), str(stack_name)) for stack_name, region in target_regions.items()
349 }
350 unexpected = {
351 region: [
352 item for item in stacks if (str(region), str(item["name"])) not in expected_targets
353 ]
354 for region, stacks in project_stacks.items()
355 if any((str(region), str(item["name"])) not in expected_targets for item in stacks)
356 }
357 if unexpected:
358 raise RuntimeError(
359 "Project stacks outside the checkpoint target set were found: "
360 + json.dumps(unexpected, sort_keys=True)
361 )
363 present: dict[str, dict[str, Any]] = {}
364 for stack_name, expected_region in target_regions.items():
365 region = str(expected_region)
366 stack = describe_stack(ctx.session, region, stack_name)
367 if stack is None or stack.get("status") == "DELETE_COMPLETE":
368 continue
369 present.setdefault(region, {})[stack_name] = _record_stack_identity(
370 ctx, stack_name, region, stack
371 )
373 checkpointed = _owned_stacks(ctx)
374 for region, records in checkpointed.items():
375 for stack_name, record in records.items():
376 if target_regions.get(stack_name) != region:
377 raise RuntimeError(
378 f"Checkpoint owns unexpected stack identity {region}:{stack_name}"
379 )
380 if str(record.get("region")) != region:
381 raise RuntimeError(f"Checkpoint Region changed for stack {region}:{stack_name}")
382 return present
385def _authorize_owned_stack(
386 ctx: RunContext,
387 stack_name: str,
388 region: str,
389 stack_id: str,
390) -> None:
391 """Revalidate checkpoint ARN and run tag at a destructive boundary."""
392 record = _owned_stack_record(ctx, region, stack_name)
393 if record is None:
394 raise RuntimeError(f"No checkpointed ownership exists for {region}:{stack_name}")
395 _require_stack_authority(
396 ctx,
397 record,
398 region=region,
399 stack_name=stack_name,
400 )
401 if str(record.get("region")) != region or str(record.get("stack_id")) != stack_id:
402 raise RuntimeError(f"Checkpoint identity changed for {region}:{stack_name}")
403 live = describe_stack(ctx.session, region, stack_id)
404 if live is None:
405 raise RuntimeError(f"Checkpointed stack disappeared before authorization: {stack_id}")
406 if live.get("name") != stack_name or live.get("stack_id") != stack_id:
407 raise RuntimeError(f"CloudFormation identity changed for {region}:{stack_name}")
408 if (live.get("tags") or {}).get(_RUN_STACK_TAG) != ctx.settings.run_id:
409 raise RuntimeError(f"Run ownership changed for {region}:{stack_name}")
412def _resolve_target_stack(
413 ctx: RunContext,
414 *,
415 region: str,
416 stack_name: str,
417 expected_stack_id: str,
418) -> dict[str, Any]:
419 """Resolve live/absent/tombstone/replacement state for one exact target."""
420 exact = describe_stack(ctx.session, region, expected_stack_id) if expected_stack_id else None
421 if exact is not None and exact.get("status") != "DELETE_COMPLETE":
422 if exact.get("name") != stack_name or exact.get("stack_id") != expected_stack_id:
423 raise RuntimeError(f"Exact stack identity changed for {region}:{stack_name}")
424 return {"state": "live", "stack": exact}
426 by_name = describe_stack(ctx.session, region, stack_name)
427 if by_name is None or by_name.get("status") == "DELETE_COMPLETE":
428 return {
429 "state": "absent",
430 "tombstone": exact if exact and exact.get("status") == "DELETE_COMPLETE" else None,
431 }
432 actual_id = str(by_name.get("stack_id") or "")
433 if expected_stack_id and actual_id != expected_stack_id:
434 return {"state": "replacement", "stack": by_name}
435 if not expected_stack_id:
436 return {"state": "uncheckpointed", "stack": by_name}
437 return {"state": "live", "stack": by_name}
440def _verify_target_stack_absence(ctx: RunContext) -> dict[str, Any]:
441 """Prove every target is absent while surfacing same-name replacements."""
442 targets = ctx.checkpoint.state.get("target_stack_regions") or {}
443 if not targets:
444 raise RuntimeError("Checkpoint lacks target stack Regions for absence verification")
445 residual: list[dict[str, Any]] = []
446 absent: list[dict[str, str]] = []
447 for stack_name, raw_region in targets.items():
448 region = str(raw_region)
449 record = _owned_stack_record(ctx, region, stack_name)
450 expected_id = str((record or {}).get("stack_id") or "")
451 resolution = _resolve_target_stack(
452 ctx,
453 region=region,
454 stack_name=stack_name,
455 expected_stack_id=expected_id,
456 )
457 if resolution["state"] == "absent":
458 absent.append({"name": stack_name, "region": region, "stack_id": expected_id})
459 continue
460 stack = resolution["stack"]
461 residual.append(
462 {
463 "name": stack_name,
464 "region": region,
465 "expected_stack_id": expected_id or None,
466 "actual_stack_id": stack.get("stack_id"),
467 "status": stack.get("status"),
468 "kind": resolution["state"],
469 }
470 )
471 return {"all_absent": not residual, "absent": absent, "residual": residual}
474def _stack_creation_time(ctx: RunContext, region: str, stack_id: str) -> datetime:
475 """CloudFormation's creation time for one exact stack ID."""
476 client = ctx.session.client("cloudformation", region_name=region)
477 stacks = client.describe_stacks(StackName=stack_id).get("Stacks", [])
478 created = stacks[0].get("CreationTime") if len(stacks) == 1 else None
479 if not isinstance(created, datetime) or created.tzinfo is None:
480 raise RuntimeError(f"CloudFormation omitted the creation time of {stack_id}")
481 return created.astimezone(UTC)
484def _adopt_run_tagged_stacks(
485 ctx: RunContext,
486 *,
487 phase: str,
488 window_started_at: datetime,
489 replaceable: Collection[str] = (),
490) -> dict[str, Any]:
491 """Adopt the target stacks a ``gco`` subprocess created or replaced for this run.
493 The deliberate relaxation of prepared-change-set authority, open only to a
494 harness whose settings opt in. A live target stack is adopted when it
495 carries this run's exact tag and CloudFormation reports it created no
496 earlier than ``window_started_at``, the server time checkpointed before
497 the phase's command started. A stack this run already owns under the same
498 ID is left as it is. One owned under a different ID is a replacement,
499 accepted only for a ``replaceable`` name: the previous generation must be
500 deleted, and it is kept in ``replaced_generations`` so what it retained
501 (its EKS key, its log groups) stays attributable to the run. Anything else
502 (a missing tag, an earlier creation time, a replaced control-plane stack)
503 fails closed and adopts nothing further.
504 """
505 if not _run_tag_adoption_allowed(ctx):
506 raise RuntimeError("Run-tag stack adoption is not enabled for this harness")
507 if window_started_at.tzinfo is None:
508 raise ValueError("The adoption window must start at a timezone-aware time")
509 target_regions = ctx.checkpoint.state.get("target_stack_regions")
510 if not isinstance(target_regions, dict) or not target_regions:
511 raise RuntimeError("Checkpoint lacks target stack Regions for run-tag adoption")
512 unknown = sorted(set(replaceable) - set(target_regions))
513 if unknown:
514 raise RuntimeError("Replaceable stacks are not targets: " + ", ".join(unknown))
515 window = window_started_at.astimezone(UTC)
516 evidence: dict[str, Any] = {
517 "phase": phase,
518 "window_started_at": window.isoformat(),
519 "adopted": [],
520 "replaced": [],
521 "unchanged": [],
522 "absent": [],
523 }
524 for stack_name, raw_region in sorted(target_regions.items()):
525 region = str(raw_region)
526 record = _owned_stack_record(ctx, region, stack_name)
527 live = describe_stack(ctx.session, region, stack_name)
528 if live is None or live.get("status") == "DELETE_COMPLETE":
529 evidence["absent"].append(
530 {
531 "name": stack_name,
532 "region": region,
533 "owned_stack_id": (record or {}).get("stack_id"),
534 }
535 )
536 continue
537 partition = ctx.session.get_partition_for_region(region)
538 if not partition:
539 raise RuntimeError(f"Could not resolve AWS partition for {region}")
540 stack_id = str(live.get("stack_id") or "")
541 prefix = (
542 f"arn:{partition}:cloudformation:{region}:{ctx.settings.expected_account}:"
543 f"stack/{stack_name}/"
544 )
545 if live.get("name") != stack_name or not stack_id.startswith(prefix):
546 raise RuntimeError(
547 f"CloudFormation returned an invalid identity for {region}:{stack_name}"
548 )
549 if (live.get("tags") or {}).get(_RUN_STACK_TAG) != ctx.settings.run_id:
550 raise RuntimeError(
551 f"Stack {region}:{stack_name} is not tagged for run {ctx.settings.run_id!r}; "
552 "refusing adoption"
553 )
554 summary = {"name": stack_name, "region": region, "stack_id": stack_id}
555 if record is not None and record.get("stack_id") == stack_id:
556 _require_stack_authority(ctx, record, region=region, stack_name=stack_name)
557 evidence["unchanged"].append({**summary, "status": live.get("status")})
558 continue
559 if record is not None and stack_name not in replaceable:
560 raise RuntimeError(
561 f"Stack {region}:{stack_name} changed identity during {phase}, which may not "
562 "replace it; refusing adoption"
563 )
564 created = _stack_creation_time(ctx, region, stack_id)
565 if created + _ADOPTION_CLOCK_RESOLUTION < window:
566 raise RuntimeError(
567 f"Stack {region}:{stack_name} was created at {created.isoformat()}, before "
568 f"{phase} started at {window.isoformat()}; refusing adoption"
569 )
570 generations: list[dict[str, Any]] = []
571 if record is not None:
572 _require_stack_authority(ctx, record, region=region, stack_name=stack_name)
573 previous_id = str(record.get("stack_id") or "")
574 previous = describe_stack(ctx.session, region, previous_id)
575 if previous is not None and previous.get("status") != "DELETE_COMPLETE":
576 raise RuntimeError(
577 f"Stack {region}:{stack_name} has two live generations; refusing adoption"
578 )
579 generations = [
580 *copy.deepcopy(record.get("replaced_generations") or []),
581 {
582 "stack_id": previous_id,
583 "authority": record.get("authority"),
584 "adoption": copy.deepcopy(record.get("adoption")),
585 "replaced_by": stack_id,
586 "replaced_during": phase,
587 "observed_status": (previous or {}).get("status", "absent"),
588 },
589 ]
590 with ctx.state_lock:
591 _owned_stacks(ctx).setdefault(region, {})[stack_name] = {
592 **summary,
593 "run_tag": ctx.settings.run_id,
594 "authority": _RUN_TAG_ADOPTION_AUTHORITY,
595 "adoption": {
596 "phase": phase,
597 "window_started_at": window.isoformat(),
598 "creation_time": created.isoformat(),
599 "adopted_at": utc_now(),
600 "status": live.get("status"),
601 },
602 "replaced_generations": generations,
603 }
604 ctx.persist_callback(ctx.checkpoint)
605 key = "replaced" if record is not None else "adopted"
606 evidence[key].append({**summary, "created_at": created.isoformat()})
607 return evidence