Coverage for gco/bedrock.py: 92.56%

1"""Shared Bedrock defaults loaded from the canonical ``cdk.json`` context.""" 

2 

3from __future__ import annotations 

4 

5import json 

6import re 

7from collections.abc import Mapping 

8from dataclasses import dataclass 

9from importlib import metadata 

10from pathlib import Path 

11from typing import TYPE_CHECKING, Any 

12 

13_BEDROCK_CONTEXT_KEY = "bedrock" 

14_DEFAULT_MODEL_ID_KEY = "default_model_id" 

15_THINKING_KEY = "thinking" 

16_THINKING_EFFORT_KEY = "effort" 

17# Effort levels accepted in ``cdk.json``. This is deliberately the 

18# *intersection* of what the supported reasoning dialects accept: Nova 2 tops 

19# out at ``high``, and while Claude Opus 4.6/5 also accept ``xhigh`` and 

20# ``max``, allowing them here would let a config value that is valid for one 

21# default model become a hard ValidationException the moment the default moves 

22# to another family. 

23_SUPPORTED_THINKING_EFFORTS = frozenset({"low", "medium", "high"}) 

24_NOVA_2_MODEL_ID_RE = re.compile(r"(?:^|/)(?:[a-z0-9-]+\.)?amazon\.nova-2-[a-z0-9-]+-v\d+:\d+$") 

25# Nova 2 rejects these three at ``high`` effort only (lower efforts keep them). 

26_NOVA_HIGH_EFFORT_UNSUPPORTED_FIELDS = frozenset({"maxTokens", "temperature", "topP"}) 

27# Strip the geography scope from a system-defined inference-profile id so the 

28# adaptive-thinking allowlist below is written once per model line rather than 

29# once per (geography, model) pair. 

30_INFERENCE_PROFILE_GEO_PREFIX_RE = re.compile(r"^(?:global|us|us-gov|eu|apac|jp|au|ca|sa|il|mx)\.") 

31# Claude model lines that accept ``thinking.type = "adaptive"``. Enumerated 

32# rather than pattern-matched because the distinction is not inferable from the 

33# id: Opus/Sonnet 4.6+ and the Mythos/Fable lines take adaptive thinking, while 

34# older Claude models (Sonnet 4.5, Opus 4.5, ...) require the legacy 

35# ``enabled`` + ``budget_tokens`` form and reject ``adaptive`` outright. An 

36# unlisted model therefore falls through to "no reasoning translation", which 

37# is the safe default rather than a guessed request shape. 

38# Source: https://docs.aws.amazon.com/bedrock/latest/userguide/claude-messages-adaptive-thinking.html 

39_CLAUDE_ADAPTIVE_THINKING_MODELS = frozenset( 

40 { 

41 "anthropic.claude-opus-5", 

42 "anthropic.claude-mythos-5", 

43 "anthropic.claude-fable-5", 

44 "anthropic.claude-opus-4-7", 

45 "anthropic.claude-mythos-preview", 

46 "anthropic.claude-opus-4-6-v1", 

47 "anthropic.claude-sonnet-4-6", 

48 } 

49) 

50# Claude dropped sampling controls starting with Opus 4.7 ("temperature, 

51# top_p, and top_k parameters are no longer supported"); verified live against 

52# the Opus 5 global profile, which answers a ValidationException for each. 

53# ``maxTokens`` is accepted and passes through, but only when a caller opts 

54# into a cap; GCO's own call sites deliberately set none, so the Converse 

55# default — the model's own maximum output length — applies. 

56_CLAUDE_UNSUPPORTED_SAMPLING_FIELDS = frozenset({"temperature", "topP", "topK"}) 

57BEDROCK_READ_TIMEOUT_SECONDS = 3600 

58_DISTRIBUTION_NAME = "gco-cli" 

59_SOURCE_ROOT = Path(__file__).resolve().parent.parent 

60_SOURCE_CDK_JSON = _SOURCE_ROOT / "cdk.json" 

61_SOURCE_CHECKOUT_MARKERS = (_SOURCE_ROOT / "app.py", _SOURCE_ROOT / "pyproject.toml") 

62_INSTALLED_DATA_PARTS = ("share", "gco", "cdk.json") 

63 

64if TYPE_CHECKING: 

65 # Runtime access is provided lazily by ``__getattr__`` below. 

66 DEFAULT_BEDROCK_MODEL_ID: str 

67 

68 

69class BedrockModelConfigurationError(RuntimeError): 

70 """The shared Bedrock model default could not be resolved safely.""" 

71 

72 

73@dataclass(frozen=True) 

74class BedrockDefaultConfiguration: 

75 """Validated canonical Bedrock model and reasoning preferences.""" 

76 

77 model_id: str 

78 thinking_effort: str 

79 

80 

81def _source_cdk_json_path() -> Path | None: 

82 """Return the checkout-owned config path, never an ambient ancestor file.""" 

83 if all(marker.is_file() for marker in _SOURCE_CHECKOUT_MARKERS): 

84 # Return the expected path even when it is missing so selection is 

85 # fail-closed instead of falling through to an older installed copy. 

86 return _SOURCE_CDK_JSON 

87 return None 

88 

89 

90def _installed_cdk_json_path() -> Path | None: 

91 """Locate installed data through this distribution's recorded file list. 

92 

93 ``setuptools`` data files follow the installer's selected scheme, which may 

94 differ from the interpreter's default ``sysconfig`` scheme for ``--user``, 

95 ``--prefix``, ``--target``, pipx, or uvx installs. Distribution metadata 

96 records the actual relocated path and therefore remains authoritative. 

97 """ 

98 try: 

99 distribution = metadata.distribution(_DISTRIBUTION_NAME) 

100 files = distribution.files or () 

101 for relative_path in files: 

102 if tuple(relative_path.parts[-3:]) == _INSTALLED_DATA_PARTS: 

103 return Path(str(distribution.locate_file(relative_path))).resolve() 

104 except metadata.PackageNotFoundError: 

105 return None 

106 except Exception as exc: 

107 raise BedrockModelConfigurationError( 

108 f"Unable to inspect installed {_DISTRIBUTION_NAME} package data: {exc}" 

109 ) from exc 

110 return None 

111 

112 

113def _canonical_cdk_json_path() -> Path: 

114 """Select exactly one checkout-owned or distribution-owned config file.""" 

115 source_path = _source_cdk_json_path() 

116 if source_path is not None: 

117 return source_path.resolve() 

118 

119 installed_path = _installed_cdk_json_path() 

120 if installed_path is not None: 

121 return installed_path 

122 

123 raise BedrockModelConfigurationError( 

124 "Could not locate canonical cdk.json in a GCO source checkout or the " 

125 f"installed {_DISTRIBUTION_NAME} distribution" 

126 ) 

127 

128 

129def _bedrock_configuration_from_payload( 

130 payload: Any, 

131 path: Path, 

132) -> BedrockDefaultConfiguration: 

133 """Extract and strictly validate the canonical Bedrock configuration.""" 

134 if not isinstance(payload, dict): 

135 raise BedrockModelConfigurationError(f"{path}: document root must be an object") 

136 

137 context = payload.get("context") 

138 if not isinstance(context, dict): 

139 raise BedrockModelConfigurationError(f"{path}: context must be an object") 

140 

141 bedrock = context.get(_BEDROCK_CONTEXT_KEY) 

142 if not isinstance(bedrock, dict): 

143 raise BedrockModelConfigurationError( 

144 f"{path}: context.{_BEDROCK_CONTEXT_KEY} must be an object" 

145 ) 

146 

147 model_id = bedrock.get(_DEFAULT_MODEL_ID_KEY) 

148 if not isinstance(model_id, str) or not model_id.strip(): 

149 raise BedrockModelConfigurationError( 

150 f"{path}: context.{_BEDROCK_CONTEXT_KEY}.{_DEFAULT_MODEL_ID_KEY} " 

151 "must be a non-empty string" 

152 ) 

153 

154 thinking = bedrock.get(_THINKING_KEY) 

155 thinking_path = f"context.{_BEDROCK_CONTEXT_KEY}.{_THINKING_KEY}" 

156 if not isinstance(thinking, dict): 

157 raise BedrockModelConfigurationError(f"{path}: {thinking_path} must be an object") 

158 if set(thinking) != {_THINKING_EFFORT_KEY}: 

159 raise BedrockModelConfigurationError( 

160 f"{path}: {thinking_path} must contain only {_THINKING_EFFORT_KEY!r}" 

161 ) 

162 

163 effort = thinking.get(_THINKING_EFFORT_KEY) 

164 if not isinstance(effort, str) or effort not in _SUPPORTED_THINKING_EFFORTS: 

165 supported = ", ".join(sorted(_SUPPORTED_THINKING_EFFORTS)) 

166 raise BedrockModelConfigurationError( 

167 f"{path}: {thinking_path}.{_THINKING_EFFORT_KEY} must be one of {supported}" 

168 ) 

169 

170 return BedrockDefaultConfiguration( 

171 model_id=model_id.strip(), 

172 thinking_effort=effort, 

173 ) 

174 

175 

176def _model_id_from_payload(payload: Any, path: Path) -> str: 

177 """Extract the model id through the full canonical validation contract.""" 

178 return _bedrock_configuration_from_payload(payload, path).model_id 

179 

180 

181def get_default_bedrock_configuration( 

182 cdk_json_path: Path | None = None, 

183) -> BedrockDefaultConfiguration: 

184 """Return the validated canonical Bedrock configuration from ``cdk.json``. 

185 

186 An explicit path is strict. Without one, resolution uses only the config 

187 owned by this GCO source checkout or the config recorded in the installed 

188 ``gco-cli`` distribution. Current-working-directory and ancestor files are 

189 deliberately ignored so an unrelated project cannot change model routing. 

190 Once selected, a missing, unreadable, malformed, or incomplete canonical 

191 file fails closed rather than falling through to a stale copy. 

192 """ 

193 path = cdk_json_path.resolve() if cdk_json_path is not None else _canonical_cdk_json_path() 

194 if not path.is_file(): 

195 raise BedrockModelConfigurationError(f"Canonical Bedrock config is not a file: {path}") 

196 

197 try: 

198 raw_payload = path.read_text(encoding="utf-8") 

199 except (OSError, UnicodeError) as exc: 

200 raise BedrockModelConfigurationError(f"Unable to read {path}: {exc}") from exc 

201 

202 try: 

203 payload = json.loads(raw_payload) 

204 except json.JSONDecodeError as exc: 

205 raise BedrockModelConfigurationError(f"Invalid JSON in {path}: {exc}") from exc 

206 

207 return _bedrock_configuration_from_payload(payload, path) 

208 

209 

210def get_default_bedrock_model_id(cdk_json_path: Path | None = None) -> str: 

211 """Return the sole checked-in Bedrock model default from ``cdk.json``.""" 

212 return get_default_bedrock_configuration(cdk_json_path).model_id 

213 

214 

215def get_default_bedrock_thinking_effort(cdk_json_path: Path | None = None) -> str: 

216 """Return the canonical default model's validated reasoning effort.""" 

217 return get_default_bedrock_configuration(cdk_json_path).thinking_effort 

218 

219 

220def _supports_nova_2_reasoning(model_id: str) -> bool: 

221 """Return whether a model/profile identifier accepts Nova 2 reasoningConfig.""" 

222 return _NOVA_2_MODEL_ID_RE.search(model_id) is not None 

223 

224 

225def _supports_claude_adaptive_thinking(model_id: str) -> bool: 

226 """Return whether the identifier names a Claude line taking adaptive thinking.""" 

227 base = _INFERENCE_PROFILE_GEO_PREFIX_RE.sub("", model_id.rsplit("/", 1)[-1]) 

228 return base in _CLAUDE_ADAPTIVE_THINKING_MODELS 

229 

230 

231def _nova_reasoning_options( 

232 inference_config: dict[str, Any], 

233 effort: str, 

234) -> dict[str, Any]: 

235 """Translate the canonical effort into Nova 2 ``reasoningConfig`` fields.""" 

236 resolved = inference_config 

237 if effort == "high": 237 ↛ 243line 237 didn't jump to line 243 because the condition on line 237 was always true

238 resolved = { 

239 key: value 

240 for key, value in resolved.items() 

241 if key not in _NOVA_HIGH_EFFORT_UNSUPPORTED_FIELDS 

242 } 

243 options: dict[str, Any] = {} 

244 if resolved: 244 ↛ 245line 244 didn't jump to line 245 because the condition on line 244 was never true

245 options["inferenceConfig"] = resolved 

246 options["additionalModelRequestFields"] = { 

247 "reasoningConfig": {"type": "enabled", "maxReasoningEffort": effort} 

248 } 

249 return options 

250 

251 

252def _claude_reasoning_options( 

253 inference_config: dict[str, Any], 

254 effort: str, 

255) -> dict[str, Any]: 

256 """Translate the canonical effort into Claude adaptive-thinking fields. 

257 

258 ``effort`` must ride in its own ``output_config`` object; Bedrock answers a 

259 ValidationException when it is nested inside ``thinking``. Unsupported 

260 sampling controls are dropped at every effort level because their removal 

261 is a model-wide change, not an effort-dependent one. 

262 """ 

263 resolved = { 

264 key: value 

265 for key, value in inference_config.items() 

266 if key not in _CLAUDE_UNSUPPORTED_SAMPLING_FIELDS 

267 } 

268 options: dict[str, Any] = {} 

269 if resolved: 

270 options["inferenceConfig"] = resolved 

271 options["additionalModelRequestFields"] = { 

272 "thinking": {"type": "adaptive"}, 

273 "output_config": {"effort": effort}, 

274 } 

275 return options 

276 

277 

278def build_bedrock_converse_options( 

279 model_id: str, 

280 *, 

281 inference_config: Mapping[str, Any] | None = None, 

282 cdk_json_path: Path | None = None, 

283 apply_default_reasoning: bool | None = None, 

284) -> dict[str, Any]: 

285 """Build model-safe optional kwargs for ``bedrock-runtime:Converse``. 

286 

287 Canonical reasoning preferences apply only when ``model_id`` is the 

288 configured default. Explicit third-party or other-model overrides retain 

289 their caller-provided inference controls and never receive model-specific 

290 reasoning fields. Callers that know whether the model was defaulted should 

291 pass ``apply_default_reasoning``; an explicit override then remains 

292 independent of canonical configuration even when its model ID happens to 

293 match the default. With no provenance flag, model-ID equality preserves the 

294 compatibility behavior. 

295 

296 Two reasoning dialects are translated, selected from the model id: 

297 

298 * Claude adaptive thinking — ``thinking.type = "adaptive"`` plus the effort 

299 in its own ``output_config`` object. ``temperature``, ``topP``, and 

300 ``topK`` are dropped because Claude removed them from Opus 4.7 onward. 

301 * Nova 2 ``reasoningConfig`` — ``maxReasoningEffort``, with ``maxTokens``, 

302 ``temperature``, and ``topP`` dropped at ``high`` effort only. 

303 

304 A default model in neither dialect keeps its caller-supplied inference 

305 controls and receives no reasoning fields. 

306 """ 

307 resolved_inference = dict(inference_config or {}) 

308 inference_only = {"inferenceConfig": resolved_inference} if resolved_inference else {} 

309 if apply_default_reasoning is False: 

310 return inference_only 

311 

312 if _supports_claude_adaptive_thinking(model_id): 

313 translate = _claude_reasoning_options 

314 elif _supports_nova_2_reasoning(model_id): 

315 translate = _nova_reasoning_options 

316 else: 

317 return inference_only 

318 

319 configuration = get_default_bedrock_configuration(cdk_json_path) 

320 

321 if model_id != configuration.model_id: 321 ↛ 322line 321 didn't jump to line 322 because the condition on line 321 was never true

322 if apply_default_reasoning is True: 

323 raise BedrockModelConfigurationError( 

324 "Default Bedrock model changed while building its Converse request" 

325 ) 

326 return inference_only 

327 

328 return translate(resolved_inference, configuration.thinking_effort) 

329 

330 

331#: Bedrock error code returned (with HTTP 404) when the account has never 

332#: submitted the Anthropic first-time-use case form. Anthropic models are gated 

333#: behind it; first-party models are not. 

334BEDROCK_FTU_FORM_ERROR_CODE = "FTUFormNotFilled" 

335 

336#: Remediation shown when an Anthropic model is invoked before the one-time 

337#: use-case form has been submitted. Deliberately names both paths: the console 

338#: is the usual route, the API is what automation needs. 

339BEDROCK_FTU_REMEDIATION = ( 

340 "Amazon Bedrock rejected the request because this AWS account has not " 

341 "submitted Anthropic's one-time first-time-use (FTU) case form, which is " 

342 "required before any Anthropic model can be invoked.\n" 

343 "Submit it once per account (or organization) either way:\n" 

344 " - Console: Amazon Bedrock > Model access > request access to the " 

345 "Anthropic model and complete the use case details form.\n" 

346 " - CLI: aws bedrock put-use-case-for-model-access " 

347 "--form-data <base64-encoded-json>\n" 

348 "See https://docs.aws.amazon.com/bedrock/latest/userguide/model-access.html " 

349 "for the form fields. Alternatively, point GCO at a model that needs no FTU " 

350 "form (for example an Amazon Nova profile) with --model, " 

351 "GCO_MISSION_BEDROCK_MODEL_ID, or cdk.json context.bedrock.default_model_id." 

352) 

353 

354 

355class BedrockFTUFormNotAcceptedError(RuntimeError): 

356 """Anthropic's one-time first-time-use case form has not been submitted. 

357 

358 Deliberately **not** a transport error. Every advisory Bedrock path in GCO 

359 degrades gracefully when a model is briefly unreachable — throttling, a 

360 dropped connection, a malformed response — because retrying or falling back 

361 to deterministic templates is the right answer for a transient fault. A 

362 missing FTU form is the opposite: it is a permanent, account-scoped 

363 misconfiguration that fails every subsequent call identically, so a silent 

364 fallback would quietly downgrade an entire Mission run (or hand back a 

365 template-derived answer) while hiding a one-line fix. This type therefore 

366 propagates through the fallback handlers and surfaces the remediation. 

367 

368 It subclasses ``RuntimeError`` so existing callers that catch ``RuntimeError`` 

369 around the capacity advisor keep working. 

370 """ 

371 

372 def __init__(self, message: str | None = None) -> None: 

373 super().__init__(message or BEDROCK_FTU_REMEDIATION) 

374 

375 

376def raise_if_bedrock_ftu_form_error(error: BaseException) -> None: 

377 """Convert an FTU-gated Bedrock failure into a hard, actionable error. 

378 

379 Call this at the top of a ``ClientError`` handler that would otherwise 

380 degrade gracefully, so the FTU case is escalated instead of absorbed. 

381 Non-FTU errors return without raising, leaving the caller's own handling 

382 untouched. 

383 """ 

384 if is_bedrock_ftu_form_error(error): 

385 raise BedrockFTUFormNotAcceptedError() from error 

386 

387 

388def is_bedrock_ftu_form_error(error: BaseException | None) -> bool: 

389 """Return whether ``error`` (or anything it was raised from) is the FTU gate. 

390 

391 The exception chain is walked because the capacity advisor re-raises the 

392 underlying ``ClientError`` as a ``RuntimeError``, so the CLI layer only ever 

393 sees the original code through ``__cause__``. 

394 """ 

395 seen: set[int] = set() 

396 current = error 

397 while current is not None and id(current) not in seen: 

398 seen.add(id(current)) 

399 response = getattr(current, "response", None) 

400 if isinstance(response, Mapping): 

401 error_block = response.get("Error") 

402 if ( 

403 isinstance(error_block, Mapping) 

404 and error_block.get("Code") == BEDROCK_FTU_FORM_ERROR_CODE 

405 ): 

406 return True 

407 if BEDROCK_FTU_FORM_ERROR_CODE in str(current): 407 ↛ 408line 407 didn't jump to line 408 because the condition on line 407 was never true

408 return True 

409 current = current.__cause__ 

410 return False 

411 

412 

413#: Remediation shown when a Converse response was cut off by an output-token 

414#: limit. GCO's own call sites set no ``maxTokens`` (the Converse default is 

415#: the model's maximum output length), so this fires only at the model's own 

416#: ceiling or under an explicitly configured cap. 

417BEDROCK_TRUNCATION_REMEDIATION = ( 

418 "Bedrock stopped generating before the answer was complete " 

419 '(stopReason="max_tokens"). GCO sets no output-token cap by default, so ' 

420 "the response hit either the model's own maximum output length or an " 

421 "explicitly configured maxTokens cap. Try a shorter or narrower request " 

422 "(for the capacity advisor: fewer instance types or regions), raise or " 

423 "remove any explicit maxTokens cap, or choose a model with a larger " 

424 "output window via --model." 

425) 

426 

427 

428class BedrockResponseTruncatedError(RuntimeError): 

429 """A Converse response was cut off by an output-token limit. 

430 

431 Raised instead of returning truncated text because every GCO consumer of 

432 Bedrock text does worse with a partial answer than with a clear failure: 

433 the capacity advisor would surface a confusing JSON-parse error and the 

434 Mission engine would record a silently incomplete rationale. Subclasses 

435 ``RuntimeError`` so existing callers that catch ``RuntimeError`` around 

436 Bedrock calls keep working. 

437 """ 

438 

439 def __init__(self, message: str | None = None) -> None: 

440 super().__init__(message or BEDROCK_TRUNCATION_REMEDIATION) 

441 

442 

443def extract_bedrock_converse_text(response: Mapping[str, Any]) -> str: 

444 """Return the first non-empty text block, skipping reasoning content. 

445 

446 Raises :class:`BedrockResponseTruncatedError` when the response was cut 

447 off by an output-token limit (``stopReason == "max_tokens"``); the 

448 truncation check runs first because a partial text block would otherwise 

449 be returned as if it were a complete answer. 

450 """ 

451 if response.get("stopReason") == "max_tokens": 451 ↛ 452line 451 didn't jump to line 452 because the condition on line 451 was never true

452 raise BedrockResponseTruncatedError() 

453 

454 content = response["output"]["message"]["content"] 

455 if not isinstance(content, list): 455 ↛ 456line 455 didn't jump to line 456 because the condition on line 455 was never true

456 raise TypeError("Bedrock response content must be a list") 

457 

458 for block in content: 

459 if not isinstance(block, Mapping): 459 ↛ 460line 459 didn't jump to line 460 because the condition on line 459 was never true

460 continue 

461 text = block.get("text") 

462 if isinstance(text, str) and text.strip(): 

463 return text 

464 

465 raise IndexError("Bedrock response contains no non-empty text block") 

466 

467 

468def __getattr__(name: str) -> Any: 

469 """Resolve the historical module constant only when it is requested.""" 

470 if name == "DEFAULT_BEDROCK_MODEL_ID": 

471 return get_default_bedrock_model_id() 

472 raise AttributeError(f"module {__name__!r} has no attribute {name!r}") 

473 

474 

475def __dir__() -> list[str]: 

476 """Advertise the lazy compatibility alias to introspection tools.""" 

477 return sorted({*globals(), "DEFAULT_BEDROCK_MODEL_ID"}) 

478 

479 

480__all__ = [ 

481 "DEFAULT_BEDROCK_MODEL_ID", 

482 "BEDROCK_FTU_FORM_ERROR_CODE", 

483 "BEDROCK_FTU_REMEDIATION", 

484 "BEDROCK_READ_TIMEOUT_SECONDS", 

485 "BEDROCK_TRUNCATION_REMEDIATION", 

486 "BedrockDefaultConfiguration", 

487 "BedrockFTUFormNotAcceptedError", 

488 "BedrockModelConfigurationError", 

489 "BedrockResponseTruncatedError", 

490 "build_bedrock_converse_options", 

491 "extract_bedrock_converse_text", 

492 "get_default_bedrock_configuration", 

493 "get_default_bedrock_model_id", 

494 "get_default_bedrock_thinking_effort", 

495 "is_bedrock_ftu_form_error", 

496 "raise_if_bedrock_ftu_form_error", 

497]