Coverage for cli / managed_config.py: 100.00%
287 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-14 22:07 +0000
1"""Managed deployment-config engine: validated, atomic, audited cdk.json edits.
3This module is the categorical answer to "add a CLI/MCP toggle for cdk.json
4knob X" requests (issue #221). Instead of re-implementing read/validate/write
5logic per knob, each externally manageable key registers a :class:`ManagedListKey`
6(set-semantics list) or :class:`ManagedScalarKey` (single string value) and
7every mutation flows through one engine that guarantees:
9- **Resolution**: the target is the same ``cdk.json`` the CDK CLI would use
10 (current directory upward), or an explicit caller-supplied path. Installed
11 (``uvx`` / ``pip``) distributions resolve to read-only package data — the
12 engine refuses those with an actionable message instead of half-working.
13- **Validation of the result, not the starting state**: an edit is accepted
14 iff the *resulting* configuration passes the same validators the CDK app
15 applies at synth time (``gco/stacks/constants.py``). This deliberately
16 allows repairing an already-broken config (e.g. removing a typo'd Region).
17- **Idempotency**: re-adding a present value or removing an absent one is a
18 reported no-op — no bytes are written, no timestamps churn.
19- **Atomicity**: writes go through the same tmp-file + ``os.replace`` dance
20 as the feature toggles in ``cli/stacks.py``, preserving file mode and the
21 original trailing-newline state, so a crash can never leave a torn file.
22- **Auditability**: every mutation attempt logs a structured line on the
23 ``gco.cli.managed_config`` logger; MCP exposure adds ``@audit_logged`` on
24 top of that.
26Comment keys (``_comment_*``) and key order in ``cdk.json`` are preserved
27because the engine round-trips the whole document with ``json.load`` /
28``json.dumps(indent=2)`` exactly like the existing feature-toggle writers.
29"""
31from __future__ import annotations
33import json
34import logging
35import os
36import stat
37from collections.abc import Callable, Iterator
38from contextlib import contextmanager
39from dataclasses import dataclass
40from pathlib import Path
41from typing import Any
43from gco.stacks.constants import (
44 validated_deployment_partition,
45 validated_regional_deployment_regions,
46)
48from .stacks import (
49 ConfigMutationLockError,
50 _atomic_write_bytes,
51 _find_cdk_json,
52)
53from .stacks import (
54 _config_mutation_lock as _shared_config_mutation_lock,
55)
57logger = logging.getLogger("gco.cli.managed_config")
59# Effective defaults mirror the reader contract in
60# ``gco/config/config_loader.py::get_deployment_regions`` — validation of a
61# candidate result must see the same effective document the CDK app will.
62_DEFAULT_SCALAR_REGION = "us-east-2"
63_DEFAULT_REGIONAL = ("us-east-1",)
66class ManagedConfigError(RuntimeError):
67 """A managed cdk.json edit was refused; the message says how to proceed."""
70@dataclass(frozen=True)
71class ChangeReport:
72 """Uniform result of one managed mutation (including reported no-ops)."""
74 key_id: str
75 action: str # "add" | "remove" | "set"
76 value: str
77 changed: bool
78 old: tuple[str, ...] | str
79 new: tuple[str, ...] | str
80 config_path: Path
82 @staticmethod
83 def _render(side: tuple[str, ...] | str) -> str:
84 return repr(list(side)) if isinstance(side, tuple) else repr(side)
86 def summary(self) -> str:
87 """One human line suitable for CLI output and audit trails."""
88 if not self.changed:
89 state = {
90 "add": "already present",
91 "remove": "not present",
92 "set": "already the value",
93 }[self.action]
94 return f"{self.key_id}: no change ({self.value!r} {state})"
95 return (
96 f"{self.key_id}: {self.action} {self.value!r} "
97 f"({self._render(self.old)} -> {self._render(self.new)}) in {self.config_path}"
98 )
101@dataclass(frozen=True)
102class ManagedListKey:
103 """Registry entry for a cdk.json context list managed with set semantics.
105 ``validate_result`` receives the full parsed cdk.json document and the
106 candidate value the list would hold after the edit; it must raise
107 ``ValueError`` to reject. Validating in document context lets a key
108 enforce cross-key invariants (the regional-Regions key checks the single
109 partition constraint against the global/API/monitoring scalars).
110 """
112 key_id: str # dotted id, e.g. "deployment_regions.regional"
113 container: str # context child object, e.g. "deployment_regions"
114 leaf: str # list key inside the container, e.g. "regional"
115 description: str
116 default: tuple[str, ...]
117 validate_result: Callable[[dict[str, Any], tuple[str, ...]], None]
120def _validate_regional_result(document: dict[str, Any], candidate: tuple[str, ...]) -> None:
121 """Reject a regional-Regions candidate the CDK app would refuse at synth.
123 Applies the exact synth-time validators: every entry must be an SDK-known
124 CloudFormation Region, unique, non-empty — and together with the effective
125 global/api_gateway/monitoring scalars must resolve to one AWS partition.
126 """
127 validated_regional_deployment_regions(list(candidate))
128 container = document.get("context", {}).get("deployment_regions", {})
129 if not isinstance(container, dict):
130 raise ValueError("context.deployment_regions must be a JSON object")
131 scalars = tuple(
132 container.get(scalar_key, _DEFAULT_SCALAR_REGION)
133 for scalar_key in ("global", "api_gateway", "monitoring")
134 )
135 validated_deployment_partition((*scalars, *candidate))
138@dataclass(frozen=True)
139class ManagedScalarKey:
140 """Registry entry for a single-valued cdk.json context string.
142 Same contract as :class:`ManagedListKey` with scalar semantics:
143 ``validate_result`` receives the full parsed document and the candidate
144 string the key would hold after the edit, raising ``ValueError`` to
145 reject. Setting the current value is a reported no-op.
146 """
148 key_id: str # dotted id, e.g. "deployment_regions.global"
149 container: str # context child object, e.g. "deployment_regions"
150 leaf: str # string key inside the container, e.g. "global"
151 description: str
152 default: str
153 validate_result: Callable[[dict[str, Any], str], None]
154 nested: tuple[str, ...] = ()
157def _effective_deployment_scalars(document: dict[str, Any]) -> dict[str, str]:
158 """Return the effective global/api_gateway/monitoring scalar Regions."""
159 container = document.get("context", {}).get("deployment_regions", {})
160 if not isinstance(container, dict):
161 raise ValueError("context.deployment_regions must be a JSON object")
162 return {
163 scalar_key: container.get(scalar_key, _DEFAULT_SCALAR_REGION)
164 for scalar_key in ("global", "api_gateway", "monitoring")
165 }
168def _effective_regional(document: dict[str, Any]) -> tuple[str, ...]:
169 """Return the effective workload-Region list (default when absent)."""
170 container = document.get("context", {}).get("deployment_regions", {})
171 if not isinstance(container, dict):
172 raise ValueError("context.deployment_regions must be a JSON object")
173 regional = container.get("regional", list(_DEFAULT_REGIONAL))
174 if not isinstance(regional, list):
175 raise ValueError("context.deployment_regions.regional must be a JSON array")
176 return tuple(regional)
179def _scalar_region_validator(role: str) -> Callable[[dict[str, Any], str], None]:
180 """Build a validator for one deployment-region scalar (``role``).
182 The candidate must be an SDK-known CloudFormation Region and the whole
183 resulting topology (candidate + the other scalars + the workload list)
184 must still resolve to one AWS partition — the same constraint synth
185 enforces, applied to the result.
186 """
188 def _validate(document: dict[str, Any], candidate: str) -> None:
189 scalars = _effective_deployment_scalars(document)
190 scalars[role] = candidate
191 validated_deployment_partition((*scalars.values(), *_effective_regional(document)))
193 return _validate
196def _bedrock_model_id_validator(key_id: str) -> Callable[[dict[str, Any], str], None]:
197 """Mirror the reader contract in ``gco/bedrock.py``: a non-empty string.
199 Model/inference-profile IDs are free-form by design (custom profiles,
200 marketplace models); the runtime readers only require a non-empty
201 string, so requiring more here would reject valid configurations. One
202 factory serves both Bedrock model knobs so their contracts cannot drift.
203 """
205 def _validate(document: dict[str, Any], candidate: str) -> None:
206 del document # no cross-key invariants for these knobs
207 if not candidate.strip():
208 raise ValueError(f"{key_id} must be a non-empty string")
209 if candidate != candidate.strip():
210 raise ValueError(f"{key_id} must not have leading/trailing whitespace")
212 return _validate
215_CODEX_REASONING_EFFORTS = frozenset({"minimal", "low", "medium", "high", "xhigh"})
218def _codex_reasoning_effort_validator(document: dict[str, Any], candidate: str) -> None:
219 """Mirror the canonical Codex effort/object contract in ``gco.bedrock``."""
220 if candidate not in _CODEX_REASONING_EFFORTS:
221 supported = ", ".join(sorted(_CODEX_REASONING_EFFORTS))
222 raise ValueError("bedrock.codex.reasoning_effort must be one of " + supported)
224 bedrock = document.get("context", {}).get("bedrock")
225 if bedrock is None:
226 return
227 if not isinstance(bedrock, dict):
228 raise ValueError("context.bedrock must be a JSON object")
229 codex = bedrock.get("codex")
230 if codex is None:
231 return
232 if not isinstance(codex, dict):
233 raise ValueError("context.bedrock.codex must be a JSON object")
234 unexpected = sorted(set(codex) - {"reasoning_effort"})
235 if unexpected:
236 raise ValueError(
237 "context.bedrock.codex must contain only 'reasoning_effort'; "
238 f"unexpected keys: {', '.join(unexpected)}"
239 )
242#: The managed-key registry. New knobs register here instead of growing
243#: bespoke read/validate/write code paths.
244REGIONAL_DEPLOYMENT_REGIONS = ManagedListKey(
245 key_id="deployment_regions.regional",
246 container="deployment_regions",
247 leaf="regional",
248 description="Workload Regions that receive an EKS regional stack",
249 default=_DEFAULT_REGIONAL,
250 validate_result=_validate_regional_result,
251)
253#: The three control-plane region scalars, addressable by role name.
254DEPLOYMENT_REGION_SCALARS: dict[str, ManagedScalarKey] = {
255 role: ManagedScalarKey(
256 key_id=f"deployment_regions.{role}",
257 container="deployment_regions",
258 leaf=role,
259 description=description,
260 default=_DEFAULT_SCALAR_REGION,
261 validate_result=_scalar_region_validator(role),
262 )
263 for role, description in (
264 ("global", "Region hosting partition-wide ECR/S3/DynamoDB and the SSM registry"),
265 ("api_gateway", "Region hosting the API Gateway stack"),
266 ("monitoring", "Region hosting the monitoring stack"),
267 )
268}
270MISSION_DEFAULT_MODEL = ManagedScalarKey(
271 key_id="bedrock.mission_default_model_id",
272 container="bedrock",
273 leaf="mission_default_model_id",
274 description="Bedrock model/inference-profile ID Mission sampling uses by default",
275 default="", # the reader has no fallback: it requires the key when consulted
276 validate_result=_bedrock_model_id_validator("bedrock.mission_default_model_id"),
277)
279CAPACITY_ADVISOR_DEFAULT_MODEL = ManagedScalarKey(
280 key_id="bedrock.capacity_advisor_default_model_id",
281 container="bedrock",
282 leaf="capacity_advisor_default_model_id",
283 description="Bedrock model/inference-profile ID the capacity advisor uses by default",
284 default="", # the reader has no fallback: it requires the key when consulted
285 validate_result=_bedrock_model_id_validator("bedrock.capacity_advisor_default_model_id"),
286)
288CLAUDE_CODE_DEFAULT_MODEL = ManagedScalarKey(
289 key_id="bedrock.claude_code_default_model_id",
290 container="bedrock",
291 leaf="claude_code_default_model_id",
292 description="Bedrock model/inference-profile ID gco autopilot hands to Claude Code",
293 default="", # the reader has no fallback: it requires the key when consulted
294 validate_result=_bedrock_model_id_validator("bedrock.claude_code_default_model_id"),
295)
297CODEX_DEFAULT_MODEL = ManagedScalarKey(
298 key_id="bedrock.codex_default_model_id",
299 container="bedrock",
300 leaf="codex_default_model_id",
301 description="Bedrock model/inference-profile ID gco autopilot hands to Codex",
302 default="", # the reader has no fallback: it requires the key when consulted
303 validate_result=_bedrock_model_id_validator("bedrock.codex_default_model_id"),
304)
306CODEX_REASONING_EFFORT = ManagedScalarKey(
307 key_id="bedrock.codex.reasoning_effort",
308 container="bedrock",
309 nested=("codex",),
310 leaf="reasoning_effort",
311 description="Canonical reasoning effort for the default Codex model",
312 default="", # the reader has no fallback: it requires the key when consulted
313 validate_result=_codex_reasoning_effort_validator,
314)
317def _resolve_config_path(config_path: Path | str | None) -> Path:
318 """Return the cdk.json to edit, refusing unusable resolutions early."""
319 if config_path is not None:
320 path = Path(config_path)
321 if not path.is_file():
322 raise ManagedConfigError(f"config path {path} does not exist or is not a file")
323 return path
324 found = _find_cdk_json()
325 if found is None:
326 raise ManagedConfigError(
327 "cdk.json not found in the current directory or any parent. "
328 "Run from a GCO checkout, or pass an explicit --config-path. "
329 "Installed (uvx/pip) distributions do not carry a writable "
330 "deployment config."
331 )
332 return found
335def _require_writable(path: Path) -> None:
336 """Refuse read-only targets (the installed-mode package-data case)."""
337 parent_writable = os.access(path.parent, os.W_OK)
338 file_writable = os.access(path, os.W_OK)
339 if parent_writable and file_writable:
340 return
341 raise ManagedConfigError(
342 f"{path} is not writable"
343 f"{'' if parent_writable else ' (directory is read-only)'}. "
344 "Installed (uvx/pip) distributions expose a read-only cdk.json; "
345 "run from a writable GCO checkout or pass --config-path pointing "
346 "at the deployment config you own."
347 )
350@contextmanager
351def _config_mutation_lock(path: Path) -> Iterator[None]:
352 """Translate shared-lock failures into this module's public error type."""
353 try:
354 with _shared_config_mutation_lock(path):
355 yield
356 except ConfigMutationLockError as exc:
357 raise ManagedConfigError(str(exc)) from exc
360def _load_document(path: Path) -> tuple[dict[str, Any], bytes]:
361 """Parse the target document, keeping raw bytes for faithful re-encoding."""
362 raw = path.read_bytes()
363 try:
364 document = json.loads(raw.decode("utf-8"))
365 except (UnicodeDecodeError, json.JSONDecodeError) as exc:
366 raise ManagedConfigError(f"{path} is not valid JSON: {exc}") from exc
367 if not isinstance(document, dict) or not isinstance(document.get("context"), dict):
368 raise ManagedConfigError(
369 f"{path} does not look like a GCO cdk.json (missing a 'context' object); "
370 "refusing to edit it"
371 )
372 return document, raw
375def _write_document(path: Path, document: dict[str, Any], original_raw: bytes) -> None:
376 """Serialize like the existing feature-toggle writers, atomically.
378 ``json.dumps(indent=2)`` with insertion order preserves ``_comment_*``
379 keys and their placement. ``ensure_ascii=False`` keeps the em dashes and
380 other non-ASCII characters inside those comments as UTF-8 rather than
381 rewriting them to ``\\uXXXX`` escapes — without it, adding one Region
382 rewrites every documented block in the file and buries the real change in
383 hundreds of lines of encoding churn. The original trailing-newline state is
384 kept so diffs stay minimal regardless of how the file was last formatted.
385 """
386 serialized = json.dumps(document, indent=2, ensure_ascii=False).encode("utf-8")
387 if original_raw.endswith(b"\n"):
388 serialized += b"\n"
389 _atomic_write_bytes(path, serialized, mode=stat.S_IMODE(path.stat().st_mode))
392def _current_values(document: dict[str, Any], key: ManagedListKey) -> tuple[str, ...]:
393 """Return the configured list, or the effective default when absent."""
394 container = document["context"].get(key.container)
395 if container is None:
396 return key.default
397 if not isinstance(container, dict):
398 raise ManagedConfigError(
399 f"context.{key.container} must be a JSON object, found {type(container).__name__}"
400 )
401 values = container.get(key.leaf)
402 if values is None:
403 return key.default
404 if not isinstance(values, list):
405 raise ManagedConfigError(
406 f"context.{key.container}.{key.leaf} must be a JSON array, "
407 f"found {type(values).__name__}"
408 )
409 return tuple(values)
412def _apply(
413 key: ManagedListKey,
414 action: str,
415 value: str,
416 config_path: Path | str | None,
417) -> ChangeReport:
418 """Shared add/remove core: resolve, lock, load, validate, and write."""
419 path = _resolve_config_path(config_path)
420 with _config_mutation_lock(path):
421 document, raw = _load_document(path)
422 old = _current_values(document, key)
424 if action == "add":
425 changed = value not in old
426 candidate = (*old, value) if changed else old
427 else:
428 changed = value in old
429 candidate = tuple(entry for entry in old if entry != value) if changed else old
431 if not changed:
432 report = ChangeReport(key.key_id, action, value, False, old, old, path)
433 logger.info(
434 "managed-config no-op: key=%s action=%s value=%s path=%s",
435 key.key_id,
436 action,
437 value,
438 path,
439 )
440 return report
442 try:
443 key.validate_result(document, candidate)
444 except ValueError as exc:
445 logger.warning(
446 "managed-config refused: key=%s action=%s value=%s path=%s reason=%s",
447 key.key_id,
448 action,
449 value,
450 path,
451 exc,
452 )
453 raise ManagedConfigError(f"refusing to update {key.key_id}: {exc}") from exc
455 _require_writable(path)
456 # Materialize only what this key manages; absent sibling scalars keep
457 # falling through to the reader defaults instead of being frozen into
458 # the file by an unrelated edit.
459 container = document["context"].setdefault(key.container, {})
460 container[key.leaf] = list(candidate)
461 _write_document(path, document, raw)
463 report = ChangeReport(key.key_id, action, value, True, old, candidate, path)
464 logger.info(
465 "managed-config write: key=%s action=%s value=%s old=%s new=%s path=%s",
466 key.key_id,
467 action,
468 value,
469 list(old),
470 list(candidate),
471 path,
472 )
473 return report
476def managed_list_add(
477 key: ManagedListKey, value: str, *, config_path: Path | str | None = None
478) -> ChangeReport:
479 """Add ``value`` to a managed list if absent; validated and atomic."""
480 return _apply(key, "add", value, config_path)
483def managed_list_remove(
484 key: ManagedListKey, value: str, *, config_path: Path | str | None = None
485) -> ChangeReport:
486 """Remove ``value`` from a managed list if present; validated and atomic."""
487 return _apply(key, "remove", value, config_path)
490def _scalar_container(
491 document: dict[str, Any],
492 key: ManagedScalarKey,
493 *,
494 materialize: bool = False,
495) -> dict[str, Any] | None:
496 """Resolve (and optionally create) a scalar key's nested object path."""
497 context = document["context"]
498 container = context.get(key.container)
499 path_parts = ["context", key.container]
500 if container is None:
501 if not materialize:
502 return None
503 container = {}
504 context[key.container] = container
505 if not isinstance(container, dict):
506 path = ".".join(path_parts)
507 raise ManagedConfigError(f"{path} must be a JSON object, found {type(container).__name__}")
509 for segment in key.nested:
510 path_parts.append(segment)
511 child = container.get(segment)
512 if child is None:
513 if not materialize:
514 return None
515 child = {}
516 container[segment] = child
517 if not isinstance(child, dict):
518 path = ".".join(path_parts)
519 raise ManagedConfigError(f"{path} must be a JSON object, found {type(child).__name__}")
520 container = child
521 return container
524def _current_scalar(document: dict[str, Any], key: ManagedScalarKey) -> str:
525 """Return the configured scalar, or the effective default when absent."""
526 container = _scalar_container(document, key)
527 if container is None:
528 return key.default
529 value = container.get(key.leaf)
530 if value is None:
531 return key.default
532 if not isinstance(value, str):
533 dotted_path = ".".join(("context", key.container, *key.nested, key.leaf))
534 raise ManagedConfigError(
535 f"{dotted_path} must be a JSON string, found {type(value).__name__}"
536 )
537 return value
540def managed_scalar_set(
541 key: ManagedScalarKey, value: str, *, config_path: Path | str | None = None
542) -> ChangeReport:
543 """Set a managed scalar; validated, atomic, and a no-op when unchanged."""
544 path = _resolve_config_path(config_path)
545 with _config_mutation_lock(path):
546 document, raw = _load_document(path)
547 old = _current_scalar(document, key)
549 try:
550 key.validate_result(document, value)
551 except ValueError as exc:
552 logger.warning(
553 "managed-config refused: key=%s action=set value=%s path=%s reason=%s",
554 key.key_id,
555 value,
556 path,
557 exc,
558 )
559 raise ManagedConfigError(f"refusing to update {key.key_id}: {exc}") from exc
561 if value == old:
562 report = ChangeReport(key.key_id, "set", value, False, old, old, path)
563 logger.info(
564 "managed-config no-op: key=%s action=set value=%s path=%s",
565 key.key_id,
566 value,
567 path,
568 )
569 return report
571 _require_writable(path)
572 # Materialize only the target path; all outer and sibling settings
573 # retain their current state / reader defaults.
574 container = _scalar_container(document, key, materialize=True)
575 if container is None: # pragma: no cover - materialize guarantees an object
576 raise ManagedConfigError(f"unable to materialize {key.key_id}")
577 container[key.leaf] = value
578 _write_document(path, document, raw)
580 report = ChangeReport(key.key_id, "set", value, True, old, value, path)
581 logger.info(
582 "managed-config write: key=%s action=set value=%s old=%s new=%s path=%s",
583 key.key_id,
584 value,
585 old,
586 value,
587 path,
588 )
589 return report
592# ---------------------------------------------------------------------------
593# Deployment-region veneers (domain-named entry points used by the CLI; the
594# MCP tools shell to the CLI commands, matching every other gated tool).
595# ---------------------------------------------------------------------------
598def get_deployment_regions_status(*, config_path: Path | str | None = None) -> dict[str, Any]:
599 """Return the effective deployment-region topology plus its partition.
601 Works on broken configurations too (this is the diagnosis entry point):
602 when the topology fails validation, ``partition`` is ``None`` and
603 ``partition_error`` carries the validator message.
604 """
605 path = _resolve_config_path(config_path)
606 document, _ = _load_document(path)
607 container_key = REGIONAL_DEPLOYMENT_REGIONS.container
608 container = document["context"].get(container_key)
609 if container is None:
610 container = {}
611 if not isinstance(container, dict):
612 raise ManagedConfigError(f"context.{container_key} must be a JSON object")
613 regional = _current_values(document, REGIONAL_DEPLOYMENT_REGIONS)
614 status: dict[str, Any] = {
615 "config_path": str(path),
616 "global": container.get("global", _DEFAULT_SCALAR_REGION),
617 "api_gateway": container.get("api_gateway", _DEFAULT_SCALAR_REGION),
618 "monitoring": container.get("monitoring", _DEFAULT_SCALAR_REGION),
619 "regional": list(regional),
620 }
621 try:
622 _validate_regional_result(document, regional)
623 status["partition"] = validated_deployment_partition(
624 (status["global"], status["api_gateway"], status["monitoring"], *regional)
625 )
626 except ValueError as exc:
627 status["partition"] = None
628 status["partition_error"] = str(exc)
629 return status
632def add_deployment_region(region: str, *, config_path: Path | str | None = None) -> ChangeReport:
633 """Add a workload Region to ``deployment_regions.regional``."""
634 return managed_list_add(REGIONAL_DEPLOYMENT_REGIONS, region, config_path=config_path)
637def remove_deployment_region(region: str, *, config_path: Path | str | None = None) -> ChangeReport:
638 """Remove a workload Region from ``deployment_regions.regional``."""
639 return managed_list_remove(REGIONAL_DEPLOYMENT_REGIONS, region, config_path=config_path)
642def set_deployment_region_role(
643 role: str, region: str, *, config_path: Path | str | None = None
644) -> ChangeReport:
645 """Set one control-plane region scalar (``global``/``api_gateway``/``monitoring``)."""
646 key = DEPLOYMENT_REGION_SCALARS.get(role)
647 if key is None:
648 raise ManagedConfigError(
649 f"unknown deployment-region role {role!r}; "
650 f"expected one of {sorted(DEPLOYMENT_REGION_SCALARS)}"
651 )
652 return managed_scalar_set(key, region, config_path=config_path)
655def get_bedrock_model_status(*, config_path: Path | str | None = None) -> dict[str, Any]:
656 """Return every managed Bedrock model/reasoning default and its path."""
657 path = _resolve_config_path(config_path)
658 document, _ = _load_document(path)
659 return {
660 "config_path": str(path),
661 "mission_default_model_id": _current_scalar(document, MISSION_DEFAULT_MODEL),
662 "capacity_advisor_default_model_id": _current_scalar(
663 document, CAPACITY_ADVISOR_DEFAULT_MODEL
664 ),
665 "claude_code_default_model_id": _current_scalar(document, CLAUDE_CODE_DEFAULT_MODEL),
666 "codex_default_model_id": _current_scalar(document, CODEX_DEFAULT_MODEL),
667 "codex_reasoning_effort": _current_scalar(document, CODEX_REASONING_EFFORT),
668 }
671def set_mission_default_model(
672 model_id: str, *, config_path: Path | str | None = None
673) -> ChangeReport:
674 """Set ``bedrock.mission_default_model_id`` (Mission sampling model default)."""
675 return managed_scalar_set(MISSION_DEFAULT_MODEL, model_id, config_path=config_path)
678def set_capacity_advisor_default_model(
679 model_id: str, *, config_path: Path | str | None = None
680) -> ChangeReport:
681 """Set ``bedrock.capacity_advisor_default_model_id`` (capacity-advisor model default)."""
682 return managed_scalar_set(CAPACITY_ADVISOR_DEFAULT_MODEL, model_id, config_path=config_path)
685def set_claude_code_default_model(
686 model_id: str, *, config_path: Path | str | None = None
687) -> ChangeReport:
688 """Set ``bedrock.claude_code_default_model_id`` (Claude session model)."""
689 return managed_scalar_set(CLAUDE_CODE_DEFAULT_MODEL, model_id, config_path=config_path)
692def set_codex_default_model(
693 model_id: str, *, config_path: Path | str | None = None
694) -> ChangeReport:
695 """Set ``bedrock.codex_default_model_id`` (Codex session model)."""
696 return managed_scalar_set(CODEX_DEFAULT_MODEL, model_id, config_path=config_path)
699def set_codex_reasoning_effort(
700 reasoning_effort: str, *, config_path: Path | str | None = None
701) -> ChangeReport:
702 """Set ``bedrock.codex.reasoning_effort`` for the canonical Codex model."""
703 return managed_scalar_set(
704 CODEX_REASONING_EFFORT,
705 reasoning_effort,
706 config_path=config_path,
707 )