From 475ac399fbcff7cc66e5e3542781d002e95d4711 Mon Sep 17 00:00:00 2001 From: Oliver Gibbs Date: Mon, 26 Oct 2026 05:34:48 +0000 Subject: [PATCH] fix(intake): resolve fabricated agents by registry record id and sanitised name when reading build status The fabrication status check (postfab._resolve_live_statuses) passed the job row's DISPLAY name (e.g. 'Arbiter Agent') as the registry name filter value. The AgentCore Registry API rejects names containing spaces or special characters with ValidationException, so every live-status read failed and the stale snapshot was used instead. Three-path resolution: 1. registryRecordId (new field on the job row) -> GetRegistryRecord (one call, exact). The fabricator now captures the record id from store_agent_config_registry via _RegistrationRunState.agent_record_info and writes it to the COMPLETED status row. 2. registryName (new field) -> ListRegistryRecords with the stamped sanitised name (the exact name the registry record was created with). 3. Fallback for pre-fix rows: sanitise the display name with sanitize_registry_name().lower() (same function the fabricator uses to derive record names) before passing it to the filter. The raw display name is never passed to the name filter. Changes: - arbiter/fabricator/index.py: _RegistrationRunState gains agent_record_info dict; store_agent_config_registry stashes (record_id, registry_name) on both create and reuse paths; _write_fabrication_status accepts registry_record_id and registry_name; process_event's COMPLETED write threads them from the run state. - service/agent_intake_single/tools/postfab.py: _resolve_registry_status takes registry_record_id and registry_name params with the three-path resolution; _resolve_live_statuses reads the new fields from job items. - tests: 7 new test classes covering by-id, by-name, sanitised-fallback, and GetRegistryRecord-failure-to-name-fallback paths. Existing tests updated: _stub_list calls use lowercased names to match the new sanitise-and-lowercase fallback convention. Live-verified: _resolve_registry_status('Arbiter Agent', '', registry_record_id='MshBnaCcFECO') -> 'APPROVED'; by sanitised name -> 'APPROVED'. --- arbiter/fabricator/index.py | 46 ++++ .../tests/test_postfab_live_registry.py | 260 ++++++++++++++++-- service/agent_intake_single/tools/postfab.py | 73 ++++- 3 files changed, 351 insertions(+), 28 deletions(-) diff --git a/arbiter/fabricator/index.py b/arbiter/fabricator/index.py index 63cfca7..fe94780 100644 --- a/arbiter/fabricator/index.py +++ b/arbiter/fabricator/index.py @@ -122,6 +122,12 @@ def __init__(self): self.succeeded = set() self.failed = {} self.pending_approval = set() + # Maps agent_use_id -> (registry_record_id, registry_name) so the + # COMPLETED status write can stamp the exact record coordinates onto + # the job row — postfab resolves by record id (one call, exact) + # instead of passing the display name through the name filter + # (which rejects special characters). + self.agent_record_info: dict[str, tuple[str, str]] = {} _registration_run_state = None @@ -344,6 +350,8 @@ def _write_fabrication_status( registration_errors: list | None = None, org_id: str | None = None, registry_approval_status: str | None = None, + registry_record_id: str | None = None, + registry_name: str | None = None, ) -> bool: """Upsert a per-agent fabrication status row in the durable jobs table. @@ -467,6 +475,16 @@ def _write_fabrication_status( names["#registryApprovalStatus"] = "registryApprovalStatus" values[":registryApprovalStatus"] = {"S": registry_approval_status} + if registry_record_id: + set_parts.append("#registryRecordId = :registryRecordId") + names["#registryRecordId"] = "registryRecordId" + values[":registryRecordId"] = {"S": registry_record_id} + + if registry_name: + set_parts.append("#registryName = :registryName") + names["#registryName"] = "registryName" + values[":registryName"] = {"S": registry_name} + update_expression = "SET " + ", ".join(set_parts) if remove_parts: update_expression += " REMOVE " + ", ".join(remove_parts) @@ -1678,6 +1696,15 @@ def store_agent_config_registry( requested_by=requested_by, org_id=org_id, ) + # Stash record coordinates so the COMPLETED status write can + # stamp them onto the job row — postfab resolves by record id + # (one call, exact) instead of passing the display name through + # the name filter (which rejects special characters). + run_state = _registration_run_state + if run_state is not None: + run_state.agent_record_info[agent_id] = ( + existing_record_id, agent_id, + ) return True # CreateRegistryRecord does NOT accept a recordId — the service generates @@ -1730,6 +1757,10 @@ def store_agent_config_registry( # activation to a usable state happens later via the catalog. print(f"[store_agent_config_registry] SUCCESS - Registry response: {response}") + # Stash record coordinates (same rationale as the reuse path above). + run_state = _registration_run_state + if run_state is not None: + run_state.agent_record_info[agent_id] = (record_id, agent_id) return True except Exception as e: @@ -2674,6 +2705,21 @@ def store_agent_config_registry_bound( if run_state is not None and run_state.pending_approval else "approved" ), + # Stamp the registry record id and sanitised registry name onto + # the job row so postfab can resolve the agent's live registry + # status by exact record id (one GetRegistryRecord call) instead + # of passing the display name through the ListRegistryRecords name + # filter (which rejects names with spaces/special characters). + registry_record_id=( + next(iter(run_state.agent_record_info.values()), (None,))[0] + if run_state is not None and run_state.agent_record_info + else None + ), + registry_name=( + next(iter(run_state.agent_record_info.values()), (None, None))[1] + if run_state is not None and run_state.agent_record_info + else None + ), ) except FabricationDeadlineExceeded as deadline_exc: diff --git a/service/agent_intake_single/tests/test_postfab_live_registry.py b/service/agent_intake_single/tests/test_postfab_live_registry.py index 1248dbf..0870bfa 100644 --- a/service/agent_intake_single/tests/test_postfab_live_registry.py +++ b/service/agent_intake_single/tests/test_postfab_live_registry.py @@ -62,12 +62,15 @@ def _job( registry_approval_status=None, updated_at=None, org_id="org-1", + registry_record_id=None, + registry_name=None, ): """Build a DynamoDB job row with the shape the fabricator writes. Keys match arbiter/fabricator/index.py::_write_fabrication_status: orchestrationId (PK), agentUseId (SK), status, agentName, updatedAt, - orgId, registryApprovalStatus, registrationErrors, errorMessage. + orgId, registryApprovalStatus, registrationErrors, errorMessage, + registryRecordId, registryName. """ row = { "orchestrationId": "sess-1", @@ -79,6 +82,10 @@ def _job( } if registry_approval_status is not None: row["registryApprovalStatus"] = registry_approval_status + if registry_record_id is not None: + row["registryRecordId"] = registry_record_id + if registry_name is not None: + row["registryName"] = registry_name return row @@ -207,7 +214,7 @@ def test_approved_overrides_stale_pending_snapshot( # Job row says pending_approval (stale snapshot) items = [_job("AgentA", "COMPLETED", registry_approval_status="pending_approval")] # Registry says APPROVED (live truth) - _stub_list(registry_stubber, "AgentA", "APPROVED") + _stub_list(registry_stubber, "agenta", "APPROVED") result = _run_status(jobs_table, items, marker_store) @@ -233,7 +240,7 @@ def test_approved_agent_not_gated_on_activation( jobs_table.query.return_value = {"Items": items} # Stub for the activation gate's own live resolution - _stub_list(registry_stubber, "AgentA", "APPROVED") + _stub_list(registry_stubber, "agenta", "APPROVED") execute = mock.MagicMock() execute.return_value = {"intakeActivateProjectAgents": { @@ -260,7 +267,7 @@ def test_pending_approval_consistent_with_registry( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED", registry_approval_status="pending_approval")] - _stub_list(registry_stubber, "AgentA", "PENDING_APPROVAL") + _stub_list(registry_stubber, "agenta", "PENDING_APPROVAL") result = _run_status(jobs_table, items, marker_store) @@ -280,7 +287,7 @@ def test_draft_shows_not_submitted( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "DRAFT") + _stub_list(registry_stubber, "agenta", "DRAFT") result = _run_status(jobs_table, items, marker_store) @@ -295,7 +302,7 @@ def test_draft_complete_section_mentions_registry( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "DRAFT") + _stub_list(registry_stubber, "agenta", "DRAFT") result = _run_status(jobs_table, items, marker_store) @@ -314,7 +321,7 @@ def test_rejected_shows_clear_message( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "REJECTED") + _stub_list(registry_stubber, "agenta", "REJECTED") result = _run_status(jobs_table, items, marker_store) @@ -333,7 +340,7 @@ def test_deprecated_shows_clear_message( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "DEPRECATED") + _stub_list(registry_stubber, "agenta", "DEPRECATED") result = _run_status(jobs_table, items, marker_store) @@ -393,8 +400,8 @@ def test_one_approved_one_pending( _job("AgentB", "COMPLETED", registry_approval_status="pending_approval"), ] # A is now approved, B is still pending - _stub_list(registry_stubber, "AgentA", "APPROVED") - _stub_list(registry_stubber, "AgentB", "PENDING_APPROVAL") + _stub_list(registry_stubber, "agenta", "APPROVED") + _stub_list(registry_stubber, "agentb", "PENDING_APPROVAL") result = _run_status(jobs_table, items, marker_store) @@ -410,7 +417,7 @@ def test_approved_and_failed_sibling( _job("AgentA", "COMPLETED"), _job("AgentB", "FAILED"), ] - _stub_list(registry_stubber, "AgentA", "APPROVED") + _stub_list(registry_stubber, "agenta", "APPROVED") result = _run_status(jobs_table, items, marker_store) @@ -425,8 +432,8 @@ def test_draft_and_approved_together( _job("AgentA", "COMPLETED"), _job("AgentB", "COMPLETED"), ] - _stub_list(registry_stubber, "AgentA", "DRAFT") - _stub_list(registry_stubber, "AgentB", "APPROVED") + _stub_list(registry_stubber, "agenta", "DRAFT") + _stub_list(registry_stubber, "agentb", "APPROVED") result = _run_status(jobs_table, items, marker_store) @@ -449,9 +456,9 @@ def test_only_first_five_agents_resolved( ): names = [f"Agent{i}" for i in range(7)] items = [_job(n, "COMPLETED") for n in names] - # Stub only the first 5 + # Stub only the first 5 — the fallback path lowercases the name for n in names[:5]: - _stub_list(registry_stubber, n, "APPROVED") + _stub_list(registry_stubber, n.lower(), "APPROVED") result = _run_status(jobs_table, items, marker_store) @@ -475,7 +482,7 @@ def test_no_raw_enums_in_approved_output( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "APPROVED") + _stub_list(registry_stubber, "agenta", "APPROVED") result = _run_status(jobs_table, items, marker_store) @@ -488,7 +495,7 @@ def test_no_raw_enums_in_draft_output( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "DRAFT") + _stub_list(registry_stubber, "agenta", "DRAFT") result = _run_status(jobs_table, items, marker_store) @@ -501,7 +508,7 @@ def test_no_raw_enums_in_pending_output( self, jobs_table, marker_store, registry_stubber, ): items = [_job("AgentA", "COMPLETED")] - _stub_list(registry_stubber, "AgentA", "PENDING_APPROVAL") + _stub_list(registry_stubber, "agenta", "PENDING_APPROVAL") result = _run_status(jobs_table, items, marker_store) @@ -600,3 +607,220 @@ def test_new_filters_shape_accepted(self): f"ParamValidator should accept filters-based shape but got errors: " f"{report.generate_report()}" ) + + +# =========================================================================== +# Resolve by registryRecordId (GetRegistryRecord, exact) +# =========================================================================== + +def _stub_get_record(stubber, record_id, name, registry_status): + """Stub a single GetRegistryRecord call returning one record.""" + stubber.add_response( + "get_registry_record", + { + "registryArn": f"arn:aws:bedrock-agentcore:us-west-2:123456:registry/reg-test-1", + "recordArn": f"arn:aws:bedrock-agentcore:us-west-2:123456:registry/reg-test-1/record/{record_id}", + "recordId": record_id, + "name": name, + "description": f"Test agent {name}", + "recordType": "CUSTOM", + "recordVersion": "1", + "status": registry_status, + "createdAt": datetime(2026, 10, 1, tzinfo=timezone.utc), + "updatedAt": datetime(2026, 10, 25, tzinfo=timezone.utc), + }, + expected_params={ + "registryId": "reg-test-1", + "recordId": record_id, + }, + ) + + +class TestResolveByRegistryRecordId: + """When the job row carries registryRecordId, resolve via + GetRegistryRecord (one call, exact) — never ListRegistryRecords.""" + + def test_resolve_by_record_id_approved( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job( + "Arbiter Agent", "COMPLETED", + registry_record_id="MshBnaCcFECO", + registry_name="arbiter-agent", + ), + ] + _stub_get_record( + registry_stubber, "MshBnaCcFECO", "arbiter-agent", "APPROVED", + ) + + result = _run_status(jobs_table, items, marker_store) + + assert result["ok"] is True + assert result["status"] == "complete" + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["Arbiter Agent"] == "built and approved" + + def test_resolve_by_record_id_pending_approval( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job( + "Arbiter Agent", "COMPLETED", + registry_record_id="MshBnaCcFECO", + registry_name="arbiter-agent", + ), + ] + _stub_get_record( + registry_stubber, "MshBnaCcFECO", "arbiter-agent", "PENDING_APPROVAL", + ) + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "awaiting approval" in states["Arbiter Agent"] + assert result["pending_approval"] == ["Arbiter Agent"] + + +# =========================================================================== +# Resolve by registryName (ListRegistryRecords with sanitised name) +# =========================================================================== + +class TestResolveByRegistryName: + """When the job row carries registryName but no registryRecordId, + resolve via ListRegistryRecords using the stamped sanitised name.""" + + def test_resolve_by_registry_name( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job( + "Arbiter Agent", "COMPLETED", + registry_name="arbiter-agent", + ), + ] + _stub_list(registry_stubber, "arbiter-agent", "APPROVED") + + result = _run_status(jobs_table, items, marker_store) + + assert result["ok"] is True + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["Arbiter Agent"] == "built and approved" + + +# =========================================================================== +# Fallback: sanitise display name when no registryRecordId/registryName +# =========================================================================== + +class TestFallbackSanitiseDisplayName: + """When the job row has NO registryRecordId and NO registryName + (pre-fix rows), the display name is sanitised with + sanitize_registry_name before being passed to the name filter. + + Fixture copied from the LIVE job row shape: agentName is + 'Catering & Fuel Sequencing Agent' — the sanitised form is + 'Catering-Fuel-Sequencing-Agent'. + """ + + def test_display_name_sanitised_for_filter( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job("Catering & Fuel Sequencing Agent", "COMPLETED"), + ] + # The sanitised, lowercased name that sanitize_registry_name().lower() + # produces: '&' and spaces -> '-', consecutive dashes collapsed, then + # lowercased to match the fabricator's LLM naming convention. + _stub_list( + registry_stubber, + "catering-fuel-sequencing-agent", + "APPROVED", + ) + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["Catering & Fuel Sequencing Agent"] == "built and approved" + + def test_display_name_validation_exception_shows_not_confirmed( + self, jobs_table, marker_store, registry_stubber, + ): + """When the registry returns an error even for the sanitised name, + the state should show 'could not be confirmed', never 'approved'.""" + items = [ + _job("Catering & Fuel Sequencing Agent", "COMPLETED"), + ] + registry_stubber.add_client_error( + "list_registry_records", + service_error_code="ValidationException", + service_message="Invalid value for filter name", + ) + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "could not be confirmed" in states["Catering & Fuel Sequencing Agent"] + + def test_param_validator_rejects_raw_display_name(self): + """Prove that passing the raw display name 'Catering & Fuel + Sequencing Agent' through the name filter would be accepted by + botocore's ParamValidator (it validates structure, not business + rules) — but the LIVE API rejects it. This test documents why + sanitisation is needed: the ValidationException is server-side, + not client-side.""" + from botocore.validate import ParamValidator + + client = boto3.client( + "agent-registry-control", region_name="ap-southeast-2", + ) + op_model = client._service_model.operation_model("ListRegistryRecords") + # This shape is structurally valid (ParamValidator passes) but the + # LIVE server rejects the ampersand/spaces — the fix is to + # sanitise before calling. + kwargs = { + "registryId": "reg-test-1", + "filters": [ + {"name": "recordType", "values": ["CUSTOM"]}, + {"name": "name", "values": ["Catering & Fuel Sequencing Agent"]}, + ], + "maxResults": 10, + } + report = ParamValidator().validate(kwargs, op_model.input_shape) + # ParamValidator is structural only — this PASSES locally but the + # server rejects it. The test documents the gap. + assert not report.has_errors() + + +# =========================================================================== +# GetRegistryRecord fallback to name on failure +# =========================================================================== + +class TestRecordIdFallbackToName: + """When GetRegistryRecord by id fails, fall through to name lookup.""" + + def test_get_record_fails_falls_back_to_name( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job( + "Arbiter Agent", "COMPLETED", + registry_record_id="BADID", + registry_name="arbiter-agent", + ), + ] + # GetRegistryRecord fails + registry_stubber.add_client_error( + "get_registry_record", + service_error_code="ResourceNotFoundException", + expected_params={ + "registryId": "reg-test-1", + "recordId": "BADID", + }, + ) + # Falls through to ListRegistryRecords by name + _stub_list(registry_stubber, "arbiter-agent", "APPROVED") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["Arbiter Agent"] == "built and approved" diff --git a/service/agent_intake_single/tools/postfab.py b/service/agent_intake_single/tools/postfab.py index 3849393..95564c5 100644 --- a/service/agent_intake_single/tools/postfab.py +++ b/service/agent_intake_single/tools/postfab.py @@ -115,27 +115,68 @@ def _reset_registry_client_for_test() -> None: _MAX_REGISTRY_LOOKUPS = 5 -def _resolve_registry_status(agent_name: str, org_id: str) -> str | None: - """Look up the LIVE registry status for a single agent by name. +def _resolve_registry_status( + agent_name: str, + org_id: str, + registry_record_id: str | None = None, + registry_name: str | None = None, +) -> str | None: + """Look up the LIVE registry status for a single agent. + + Resolution order (most-to-least precise): + 1. ``registry_record_id`` — one GetRegistryRecord call, exact. + 2. ``registry_name`` — the sanitised record name written by the + fabricator; ListRegistryRecords with the ``name`` filter. + 3. Sanitise the display ``agent_name`` with the SAME function the + fabricator uses (``sanitize_registry_name``, imported from + ``tools.registry_name`` — the Python twin of the backend + sanitiser) and list with filters. + + Never passes the raw display name to the ``name`` filter — that is the + bug this fix addresses (the API rejects names with spaces/special + characters with ValidationException). Returns the raw registry status string (e.g. 'APPROVED', 'PENDING_APPROVAL', 'DRAFT') or None when the record cannot be found or the lookup fails. Sequential and bounded — called at most _MAX_REGISTRY_LOOKUPS times per status check. """ - if not REGISTRY_ID or not agent_name: + if not REGISTRY_ID: return None try: client = _get_registry_client() - # Use filters (not top-level name kwarg) — ListRegistryRecords has - # no top-level `name` parameter; the API accepts `filters` to narrow - # results. Mirrors arbiter/fabricator/index.py::_find_existing_record_id - # (source of truth for the filter shape). + + # Path 1: exact record id — one call, no filter ambiguity. + if registry_record_id: + try: + resp = client.get_registry_record( + registryId=REGISTRY_ID, + recordId=registry_record_id, + ) + if isinstance(resp, dict) and resp.get("status"): + return resp["status"] + except Exception as e: + logger.warning( + "registry_status_lookup_by_id_failed record_id=%s " + "error_type=%s: %s — falling through to name lookup", + registry_record_id, type(e).__name__, e, + ) + + # Path 2/3: list by name — prefer the stamped registry_name, else + # sanitise the display name with the SAME function the fabricator + # uses to create record names. The fabricator's LLM produces + # lowercase names (e.g. 'arbiter-agent'), and the registry name + # filter is case-sensitive, so the sanitised fallback is also + # lowercased to match the convention. + lookup_name = registry_name or sanitize_registry_name(agent_name).lower() + if not lookup_name: + return None + response = client.list_registry_records( registryId=REGISTRY_ID, filters=[ {"name": "recordType", "values": ["CUSTOM"]}, - {"name": "name", "values": [agent_name]}, + {"name": "name", "values": [lookup_name]}, ], maxResults=10, ) @@ -144,7 +185,7 @@ def _resolve_registry_status(agent_name: str, org_id: str) -> str | None: for summary in response.get("registryRecords", []): if not isinstance(summary, dict): continue - if summary.get("name") == agent_name: + if summary.get("name") == lookup_name: return summary.get("status") return None except Exception as e: @@ -168,6 +209,13 @@ def _resolve_live_statuses( on miss). Only the first _MAX_REGISTRY_LOOKUPS COMPLETED agents are resolved; the rest get None (fall back to job row snapshot). Reads are sequential to stay within SDK rate limits. + + Resolution priority per item (see ``_resolve_registry_status``): + 1. ``registryRecordId`` → GetRegistryRecord (one call, exact). + 2. ``registryName`` → ListRegistryRecords by sanitised name. + 3. Sanitise the display ``agentName`` with ``sanitize_registry_name`` + (same function the fabricator uses) → ListRegistryRecords. + Never passes the raw display name to the ``name`` filter. """ result: dict[str, str | None] = {} # Derive org_id from the current session for org-scoped lookups. @@ -187,7 +235,12 @@ def _resolve_live_statuses( result[name] = None continue lookups += 1 - result[name] = _resolve_registry_status(name, org_id) + result[name] = _resolve_registry_status( + name, + org_id, + registry_record_id=item.get("registryRecordId"), + registry_name=item.get("registryName"), + ) return result