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