Coverage for cli / managed_config.py: 100.00%
290 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-10-28 16:19 +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 and unique — and together with the effective
125 global/api_gateway/monitoring scalars must resolve to one AWS partition.
126 An empty candidate is valid: it describes a control-plane-only deployment
127 (global, API Gateway, and monitoring stacks with no workload cluster).
128 """
129 validated_regional_deployment_regions(list(candidate))
130 container = document.get("context", {}).get("deployment_regions", {})
131 if not isinstance(container, dict):
132 raise ValueError("context.deployment_regions must be a JSON object")
133 scalars = tuple(
134 container.get(scalar_key, _DEFAULT_SCALAR_REGION)
135 for scalar_key in ("global", "api_gateway", "monitoring")
136 )
137 validated_deployment_partition((*scalars, *candidate))
140@dataclass(frozen=True)
141class ManagedScalarKey:
142 """Registry entry for a single-valued cdk.json context string.
144 Same contract as :class:`ManagedListKey` with scalar semantics:
145 ``validate_result`` receives the full parsed document and the candidate
146 string the key would hold after the edit, raising ``ValueError`` to
147 reject. Setting the current value is a reported no-op.
148 """
150 key_id: str # dotted id, e.g. "deployment_regions.global"
151 container: str # context child object, e.g. "deployment_regions"
152 leaf: str # string key inside the container, e.g. "global"
153 description: str
154 default: str
155 validate_result: Callable[[dict[str, Any], str], None]
156 nested: tuple[str, ...] = ()
159def _effective_deployment_scalars(document: dict[str, Any]) -> dict[str, str]:
160 """Return the effective global/api_gateway/monitoring scalar Regions."""
161 container = document.get("context", {}).get("deployment_regions", {})
162 if not isinstance(container, dict):
163 raise ValueError("context.deployment_regions must be a JSON object")
164 return {
165 scalar_key: container.get(scalar_key, _DEFAULT_SCALAR_REGION)
166 for scalar_key in ("global", "api_gateway", "monitoring")
167 }
170def _effective_regional(document: dict[str, Any]) -> tuple[str, ...]:
171 """Return the effective workload-Region list (default when absent)."""
172 container = document.get("context", {}).get("deployment_regions", {})
173 if not isinstance(container, dict):
174 raise ValueError("context.deployment_regions must be a JSON object")
175 regional = container.get("regional", list(_DEFAULT_REGIONAL))
176 if not isinstance(regional, list):
177 raise ValueError("context.deployment_regions.regional must be a JSON array")
178 return tuple(regional)
181def _scalar_region_validator(role: str) -> Callable[[dict[str, Any], str], None]:
182 """Build a validator for one deployment-region scalar (``role``).
184 The candidate must be an SDK-known CloudFormation Region and the whole
185 resulting topology (candidate + the other scalars + the workload list)
186 must still resolve to one AWS partition — the same constraint synth
187 enforces, applied to the result.
188 """
190 def _validate(document: dict[str, Any], candidate: str) -> None:
191 scalars = _effective_deployment_scalars(document)
192 scalars[role] = candidate
193 validated_deployment_partition((*scalars.values(), *_effective_regional(document)))
195 return _validate
198def _bedrock_model_id_validator(key_id: str) -> Callable[[dict[str, Any], str], None]:
199 """Mirror the reader contract in ``gco/bedrock.py``: a non-empty string.
201 Model/inference-profile IDs are free-form by design (custom profiles,
202 marketplace models); the runtime readers only require a non-empty
203 string, so requiring more here would reject valid configurations. One
204 factory serves both Bedrock model knobs so their contracts cannot drift.
205 """
207 def _validate(document: dict[str, Any], candidate: str) -> None:
208 del document # no cross-key invariants for these knobs
209 if not candidate.strip():
210 raise ValueError(f"{key_id} must be a non-empty string")
211 if candidate != candidate.strip():
212 raise ValueError(f"{key_id} must not have leading/trailing whitespace")
214 return _validate
217_CODEX_REASONING_EFFORTS = frozenset({"minimal", "low", "medium", "high", "xhigh"})
220def _codex_reasoning_effort_validator(document: dict[str, Any], candidate: str) -> None:
221 """Mirror the canonical Codex effort/object contract in ``gco.bedrock``."""
222 if candidate not in _CODEX_REASONING_EFFORTS:
223 supported = ", ".join(sorted(_CODEX_REASONING_EFFORTS))
224 raise ValueError("bedrock.codex.reasoning_effort must be one of " + supported)
226 bedrock = document.get("context", {}).get("bedrock")
227 if bedrock is None:
228 return
229 if not isinstance(bedrock, dict):
230 raise ValueError("context.bedrock must be a JSON object")
231 codex = bedrock.get("codex")
232 if codex is None:
233 return
234 if not isinstance(codex, dict):
235 raise ValueError("context.bedrock.codex must be a JSON object")
236 unexpected = sorted(set(codex) - {"reasoning_effort"})
237 if unexpected:
238 raise ValueError(
239 "context.bedrock.codex must contain only 'reasoning_effort'; "
240 f"unexpected keys: {', '.join(unexpected)}"
241 )
244#: The managed-key registry. New knobs register here instead of growing
245#: bespoke read/validate/write code paths.
246REGIONAL_DEPLOYMENT_REGIONS = ManagedListKey(
247 key_id="deployment_regions.regional",
248 container="deployment_regions",
249 leaf="regional",
250 description="Workload Regions that receive an EKS regional stack",
251 default=_DEFAULT_REGIONAL,
252 validate_result=_validate_regional_result,
253)
255#: The three control-plane region scalars, addressable by role name.
256DEPLOYMENT_REGION_SCALARS: dict[str, ManagedScalarKey] = {
257 role: ManagedScalarKey(
258 key_id=f"deployment_regions.{role}",
259 container="deployment_regions",
260 leaf=role,
261 description=description,
262 default=_DEFAULT_SCALAR_REGION,
263 validate_result=_scalar_region_validator(role),
264 )
265 for role, description in (
266 ("global", "Region hosting partition-wide ECR/S3/DynamoDB and the SSM registry"),
267 ("api_gateway", "Region hosting the API Gateway stack"),
268 ("monitoring", "Region hosting the monitoring stack"),
269 )
270}
272MISSION_DEFAULT_MODEL = ManagedScalarKey(
273 key_id="bedrock.mission_default_model_id",
274 container="bedrock",
275 leaf="mission_default_model_id",
276 description="Bedrock model/inference-profile ID Mission sampling uses by default",
277 default="", # the reader has no fallback: it requires the key when consulted
278 validate_result=_bedrock_model_id_validator("bedrock.mission_default_model_id"),
279)
281CAPACITY_ADVISOR_DEFAULT_MODEL = ManagedScalarKey(
282 key_id="bedrock.capacity_advisor_default_model_id",
283 container="bedrock",
284 leaf="capacity_advisor_default_model_id",
285 description="Bedrock model/inference-profile ID the capacity advisor uses by default",
286 default="", # the reader has no fallback: it requires the key when consulted
287 validate_result=_bedrock_model_id_validator("bedrock.capacity_advisor_default_model_id"),
288)
290CLAUDE_CODE_DEFAULT_MODEL = ManagedScalarKey(
291 key_id="bedrock.claude_code_default_model_id",
292 container="bedrock",
293 leaf="claude_code_default_model_id",
294 description="Bedrock model/inference-profile ID gco autopilot hands to Claude Code",
295 default="", # the reader has no fallback: it requires the key when consulted
296 validate_result=_bedrock_model_id_validator("bedrock.claude_code_default_model_id"),
297)
299CODEX_DEFAULT_MODEL = ManagedScalarKey(
300 key_id="bedrock.codex_default_model_id",
301 container="bedrock",
302 leaf="codex_default_model_id",
303 description="Bedrock model/inference-profile ID gco autopilot hands to Codex",
304 default="", # the reader has no fallback: it requires the key when consulted
305 validate_result=_bedrock_model_id_validator("bedrock.codex_default_model_id"),
306)
308CODEX_REASONING_EFFORT = ManagedScalarKey(
309 key_id="bedrock.codex.reasoning_effort",
310 container="bedrock",
311 nested=("codex",),
312 leaf="reasoning_effort",
313 description="Canonical reasoning effort for the default Codex model",
314 default="", # the reader has no fallback: it requires the key when consulted
315 validate_result=_codex_reasoning_effort_validator,
316)
318OPENCODE_DEFAULT_MODEL = ManagedScalarKey(
319 key_id="bedrock.opencode_default_model_id",
320 container="bedrock",
321 leaf="opencode_default_model_id",
322 description="Bedrock model/inference-profile ID gco autopilot hands to OpenCode",
323 default="", # the reader has no fallback: it requires the key when consulted
324 validate_result=_bedrock_model_id_validator("bedrock.opencode_default_model_id"),
325)
328def _resolve_config_path(config_path: Path | str | None) -> Path:
329 """Return the cdk.json to edit, refusing unusable resolutions early."""
330 if config_path is not None:
331 path = Path(config_path)
332 if not path.is_file():
333 raise ManagedConfigError(f"config path {path} does not exist or is not a file")
334 return path
335 found = _find_cdk_json()
336 if found is None:
337 raise ManagedConfigError(
338 "cdk.json not found in the current directory or any parent. "
339 "Run from a GCO checkout, or pass an explicit --config-path. "
340 "Installed (uvx/pip) distributions do not carry a writable "
341 "deployment config."
342 )
343 return found
346def _require_writable(path: Path) -> None:
347 """Refuse read-only targets (the installed-mode package-data case)."""
348 parent_writable = os.access(path.parent, os.W_OK)
349 file_writable = os.access(path, os.W_OK)
350 if parent_writable and file_writable:
351 return
352 raise ManagedConfigError(
353 f"{path} is not writable"
354 f"{'' if parent_writable else ' (directory is read-only)'}. "
355 "Installed (uvx/pip) distributions expose a read-only cdk.json; "
356 "run from a writable GCO checkout or pass --config-path pointing "
357 "at the deployment config you own."
358 )
361@contextmanager
362def _config_mutation_lock(path: Path) -> Iterator[None]:
363 """Translate shared-lock failures into this module's public error type."""
364 try:
365 with _shared_config_mutation_lock(path):
366 yield
367 except ConfigMutationLockError as exc:
368 raise ManagedConfigError(str(exc)) from exc
371def _load_document(path: Path) -> tuple[dict[str, Any], bytes]:
372 """Parse the target document, keeping raw bytes for faithful re-encoding."""
373 raw = path.read_bytes()
374 try:
375 document = json.loads(raw.decode("utf-8"))
376 except (UnicodeDecodeError, json.JSONDecodeError) as exc:
377 raise ManagedConfigError(f"{path} is not valid JSON: {exc}") from exc
378 if not isinstance(document, dict) or not isinstance(document.get("context"), dict):
379 raise ManagedConfigError(
380 f"{path} does not look like a GCO cdk.json (missing a 'context' object); "
381 "refusing to edit it"
382 )
383 return document, raw
386def _write_document(path: Path, document: dict[str, Any], original_raw: bytes) -> None:
387 """Serialize like the existing feature-toggle writers, atomically.
389 ``json.dumps(indent=2)`` with insertion order preserves ``_comment_*``
390 keys and their placement. ``ensure_ascii=False`` keeps the em dashes and
391 other non-ASCII characters inside those comments as UTF-8 rather than
392 rewriting them to ``\\uXXXX`` escapes — without it, adding one Region
393 rewrites every documented block in the file and buries the real change in
394 hundreds of lines of encoding churn. The original trailing-newline state is
395 kept so diffs stay minimal regardless of how the file was last formatted.
396 """
397 serialized = json.dumps(document, indent=2, ensure_ascii=False).encode("utf-8")
398 if original_raw.endswith(b"\n"):
399 serialized += b"\n"
400 _atomic_write_bytes(path, serialized, mode=stat.S_IMODE(path.stat().st_mode))
403def _current_values(document: dict[str, Any], key: ManagedListKey) -> tuple[str, ...]:
404 """Return the configured list, or the effective default when absent."""
405 container = document["context"].get(key.container)
406 if container is None:
407 return key.default
408 if not isinstance(container, dict):
409 raise ManagedConfigError(
410 f"context.{key.container} must be a JSON object, found {type(container).__name__}"
411 )
412 values = container.get(key.leaf)
413 if values is None:
414 return key.default
415 if not isinstance(values, list):
416 raise ManagedConfigError(
417 f"context.{key.container}.{key.leaf} must be a JSON array, "
418 f"found {type(values).__name__}"
419 )
420 return tuple(values)
423def _apply(
424 key: ManagedListKey,
425 action: str,
426 value: str,
427 config_path: Path | str | None,
428) -> ChangeReport:
429 """Shared add/remove core: resolve, lock, load, validate, and write."""
430 path = _resolve_config_path(config_path)
431 with _config_mutation_lock(path):
432 document, raw = _load_document(path)
433 old = _current_values(document, key)
435 if action == "add":
436 changed = value not in old
437 candidate = (*old, value) if changed else old
438 else:
439 changed = value in old
440 candidate = tuple(entry for entry in old if entry != value) if changed else old
442 if not changed:
443 report = ChangeReport(key.key_id, action, value, False, old, old, path)
444 logger.info(
445 "managed-config no-op: key=%s action=%s value=%s path=%s",
446 key.key_id,
447 action,
448 value,
449 path,
450 )
451 return report
453 try:
454 key.validate_result(document, candidate)
455 except ValueError as exc:
456 logger.warning(
457 "managed-config refused: key=%s action=%s value=%s path=%s reason=%s",
458 key.key_id,
459 action,
460 value,
461 path,
462 exc,
463 )
464 raise ManagedConfigError(f"refusing to update {key.key_id}: {exc}") from exc
466 _require_writable(path)
467 # Materialize only what this key manages; absent sibling scalars keep
468 # falling through to the reader defaults instead of being frozen into
469 # the file by an unrelated edit.
470 container = document["context"].setdefault(key.container, {})
471 container[key.leaf] = list(candidate)
472 _write_document(path, document, raw)
474 report = ChangeReport(key.key_id, action, value, True, old, candidate, path)
475 logger.info(
476 "managed-config write: key=%s action=%s value=%s old=%s new=%s path=%s",
477 key.key_id,
478 action,
479 value,
480 list(old),
481 list(candidate),
482 path,
483 )
484 return report
487def managed_list_add(
488 key: ManagedListKey, value: str, *, config_path: Path | str | None = None
489) -> ChangeReport:
490 """Add ``value`` to a managed list if absent; validated and atomic."""
491 return _apply(key, "add", value, config_path)
494def managed_list_remove(
495 key: ManagedListKey, value: str, *, config_path: Path | str | None = None
496) -> ChangeReport:
497 """Remove ``value`` from a managed list if present; validated and atomic."""
498 return _apply(key, "remove", value, config_path)
501def _scalar_container(
502 document: dict[str, Any],
503 key: ManagedScalarKey,
504 *,
505 materialize: bool = False,
506) -> dict[str, Any] | None:
507 """Resolve (and optionally create) a scalar key's nested object path."""
508 context = document["context"]
509 container = context.get(key.container)
510 path_parts = ["context", key.container]
511 if container is None:
512 if not materialize:
513 return None
514 container = {}
515 context[key.container] = container
516 if not isinstance(container, dict):
517 path = ".".join(path_parts)
518 raise ManagedConfigError(f"{path} must be a JSON object, found {type(container).__name__}")
520 for segment in key.nested:
521 path_parts.append(segment)
522 child = container.get(segment)
523 if child is None:
524 if not materialize:
525 return None
526 child = {}
527 container[segment] = child
528 if not isinstance(child, dict):
529 path = ".".join(path_parts)
530 raise ManagedConfigError(f"{path} must be a JSON object, found {type(child).__name__}")
531 container = child
532 return container
535def _current_scalar(document: dict[str, Any], key: ManagedScalarKey) -> str:
536 """Return the configured scalar, or the effective default when absent."""
537 container = _scalar_container(document, key)
538 if container is None:
539 return key.default
540 value = container.get(key.leaf)
541 if value is None:
542 return key.default
543 if not isinstance(value, str):
544 dotted_path = ".".join(("context", key.container, *key.nested, key.leaf))
545 raise ManagedConfigError(
546 f"{dotted_path} must be a JSON string, found {type(value).__name__}"
547 )
548 return value
551def managed_scalar_set(
552 key: ManagedScalarKey, value: str, *, config_path: Path | str | None = None
553) -> ChangeReport:
554 """Set a managed scalar; validated, atomic, and a no-op when unchanged."""
555 path = _resolve_config_path(config_path)
556 with _config_mutation_lock(path):
557 document, raw = _load_document(path)
558 old = _current_scalar(document, key)
560 try:
561 key.validate_result(document, value)
562 except ValueError as exc:
563 logger.warning(
564 "managed-config refused: key=%s action=set value=%s path=%s reason=%s",
565 key.key_id,
566 value,
567 path,
568 exc,
569 )
570 raise ManagedConfigError(f"refusing to update {key.key_id}: {exc}") from exc
572 if value == old:
573 report = ChangeReport(key.key_id, "set", value, False, old, old, path)
574 logger.info(
575 "managed-config no-op: key=%s action=set value=%s path=%s",
576 key.key_id,
577 value,
578 path,
579 )
580 return report
582 _require_writable(path)
583 # Materialize only the target path; all outer and sibling settings
584 # retain their current state / reader defaults.
585 container = _scalar_container(document, key, materialize=True)
586 if container is None: # pragma: no cover - materialize guarantees an object
587 raise ManagedConfigError(f"unable to materialize {key.key_id}")
588 container[key.leaf] = value
589 _write_document(path, document, raw)
591 report = ChangeReport(key.key_id, "set", value, True, old, value, path)
592 logger.info(
593 "managed-config write: key=%s action=set value=%s old=%s new=%s path=%s",
594 key.key_id,
595 value,
596 old,
597 value,
598 path,
599 )
600 return report
603# ---------------------------------------------------------------------------
604# Deployment-region veneers (domain-named entry points used by the CLI; the
605# MCP tools shell to the CLI commands, matching every other gated tool).
606# ---------------------------------------------------------------------------
609def get_deployment_regions_status(*, config_path: Path | str | None = None) -> dict[str, Any]:
610 """Return the effective deployment-region topology plus its partition.
612 Works on broken configurations too (this is the diagnosis entry point):
613 when the topology fails validation, ``partition`` is ``None`` and
614 ``partition_error`` carries the validator message.
615 """
616 path = _resolve_config_path(config_path)
617 document, _ = _load_document(path)
618 container_key = REGIONAL_DEPLOYMENT_REGIONS.container
619 container = document["context"].get(container_key)
620 if container is None:
621 container = {}
622 if not isinstance(container, dict):
623 raise ManagedConfigError(f"context.{container_key} must be a JSON object")
624 regional = _current_values(document, REGIONAL_DEPLOYMENT_REGIONS)
625 status: dict[str, Any] = {
626 "config_path": str(path),
627 "global": container.get("global", _DEFAULT_SCALAR_REGION),
628 "api_gateway": container.get("api_gateway", _DEFAULT_SCALAR_REGION),
629 "monitoring": container.get("monitoring", _DEFAULT_SCALAR_REGION),
630 "regional": list(regional),
631 }
632 try:
633 _validate_regional_result(document, regional)
634 status["partition"] = validated_deployment_partition(
635 (status["global"], status["api_gateway"], status["monitoring"], *regional)
636 )
637 except ValueError as exc:
638 status["partition"] = None
639 status["partition_error"] = str(exc)
640 return status
643def add_deployment_region(region: str, *, config_path: Path | str | None = None) -> ChangeReport:
644 """Add a workload Region to ``deployment_regions.regional``."""
645 return managed_list_add(REGIONAL_DEPLOYMENT_REGIONS, region, config_path=config_path)
648def remove_deployment_region(region: str, *, config_path: Path | str | None = None) -> ChangeReport:
649 """Remove a workload Region from ``deployment_regions.regional``."""
650 return managed_list_remove(REGIONAL_DEPLOYMENT_REGIONS, region, config_path=config_path)
653def set_deployment_region_role(
654 role: str, region: str, *, config_path: Path | str | None = None
655) -> ChangeReport:
656 """Set one control-plane region scalar (``global``/``api_gateway``/``monitoring``)."""
657 key = DEPLOYMENT_REGION_SCALARS.get(role)
658 if key is None:
659 raise ManagedConfigError(
660 f"unknown deployment-region role {role!r}; "
661 f"expected one of {sorted(DEPLOYMENT_REGION_SCALARS)}"
662 )
663 return managed_scalar_set(key, region, config_path=config_path)
666def get_bedrock_model_status(*, config_path: Path | str | None = None) -> dict[str, Any]:
667 """Return every managed Bedrock model/reasoning default and its path."""
668 path = _resolve_config_path(config_path)
669 document, _ = _load_document(path)
670 return {
671 "config_path": str(path),
672 "mission_default_model_id": _current_scalar(document, MISSION_DEFAULT_MODEL),
673 "capacity_advisor_default_model_id": _current_scalar(
674 document, CAPACITY_ADVISOR_DEFAULT_MODEL
675 ),
676 "claude_code_default_model_id": _current_scalar(document, CLAUDE_CODE_DEFAULT_MODEL),
677 "codex_default_model_id": _current_scalar(document, CODEX_DEFAULT_MODEL),
678 "codex_reasoning_effort": _current_scalar(document, CODEX_REASONING_EFFORT),
679 "opencode_default_model_id": _current_scalar(document, OPENCODE_DEFAULT_MODEL),
680 }
683def set_mission_default_model(
684 model_id: str, *, config_path: Path | str | None = None
685) -> ChangeReport:
686 """Set ``bedrock.mission_default_model_id`` (Mission sampling model default)."""
687 return managed_scalar_set(MISSION_DEFAULT_MODEL, model_id, config_path=config_path)
690def set_capacity_advisor_default_model(
691 model_id: str, *, config_path: Path | str | None = None
692) -> ChangeReport:
693 """Set ``bedrock.capacity_advisor_default_model_id`` (capacity-advisor model default)."""
694 return managed_scalar_set(CAPACITY_ADVISOR_DEFAULT_MODEL, model_id, config_path=config_path)
697def set_claude_code_default_model(
698 model_id: str, *, config_path: Path | str | None = None
699) -> ChangeReport:
700 """Set ``bedrock.claude_code_default_model_id`` (Claude session model)."""
701 return managed_scalar_set(CLAUDE_CODE_DEFAULT_MODEL, model_id, config_path=config_path)
704def set_codex_default_model(
705 model_id: str, *, config_path: Path | str | None = None
706) -> ChangeReport:
707 """Set ``bedrock.codex_default_model_id`` (Codex session model)."""
708 return managed_scalar_set(CODEX_DEFAULT_MODEL, model_id, config_path=config_path)
711def set_codex_reasoning_effort(
712 reasoning_effort: str, *, config_path: Path | str | None = None
713) -> ChangeReport:
714 """Set ``bedrock.codex.reasoning_effort`` for the canonical Codex model."""
715 return managed_scalar_set(
716 CODEX_REASONING_EFFORT,
717 reasoning_effort,
718 config_path=config_path,
719 )
722def set_opencode_default_model(
723 model_id: str, *, config_path: Path | str | None = None
724) -> ChangeReport:
725 """Set ``bedrock.opencode_default_model_id`` (OpenCode session model)."""
726 return managed_scalar_set(OPENCODE_DEFAULT_MODEL, model_id, config_path=config_path)