diff --git a/service/agent_intake_single/tests/test_postfab_live_registry.py b/service/agent_intake_single/tests/test_postfab_live_registry.py new file mode 100644 index 0000000..4cd0fec --- /dev/null +++ b/service/agent_intake_single/tests/test_postfab_live_registry.py @@ -0,0 +1,602 @@ +"""Tests for live registry status resolution in check_fabrication_status. + +The fix replaces the write-once ``registryApprovalStatus`` snapshot on the +job row with a live query to the AgentCore Registry. For COMPLETED jobs the +postfab tool now calls ``ListRegistryRecords(filters=[{name:'recordType', +values:['CUSTOM']},{name:'name',values:[agent_name]}])`` and derives +the human-readable state from the LIVE record status: + + DRAFT -> 'built, not yet submitted for approval' + PENDING_APPROVAL -> 'awaiting approval on the Approvals page' + APPROVED -> 'built and approved' + REJECTED -> 'built, but the approval was rejected' + DEPRECATED -> 'built, but has been deprecated' + +Each test uses the botocore Stubber so kwargs are validated against the +real service model — no mocks that could drift from the API shape. + +A dedicated ``TestParamValidatorRejectsTopLevelName`` class validates the +built kwargs with ``botocore.validate.ParamValidator`` against the REAL +service model's input shape — proving the old ``name=agent_name`` top-level +kwarg would be rejected by the live API. + +Job row fixture mirrors the LIVE row shape written by +``arbiter/fabricator/index.py::_write_fabrication_status``. + +Run with: + PYTHONPATH=. pytest tests/test_postfab_live_registry.py -v --tb=short +from the service/agent_intake_single directory. +""" +import json +import os +import sys +from datetime import datetime, timezone +from unittest import mock + +import boto3 +import pytest +from botocore.stub import Stubber + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +import request_context + +os.environ.setdefault("APPSYNC_GRAPHQL_URL", "https://test.example/graphql") +os.environ.setdefault("FABRICATION_JOBS_TABLE", "jobs-test") +os.environ.setdefault("REGISTRY_ID", "reg-test-1") + +import tools.postfab as postfab + + +# --------------------------------------------------------------------------- +# Helpers — live-shape job rows +# --------------------------------------------------------------------------- + +def _fresh_ts(): + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _job( + name, + status, + registry_approval_status=None, + updated_at=None, + org_id="org-1", +): + """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. + """ + row = { + "orchestrationId": "sess-1", + "agentUseId": name, + "agentName": name, + "status": status, + "updatedAt": updated_at or _fresh_ts(), + "orgId": org_id, + } + if registry_approval_status is not None: + row["registryApprovalStatus"] = registry_approval_status + return row + + +def _registry_record(name, status, record_id=None): + """Build a registry record summary matching the ListRegistryRecords shape + on the ``agent-registry-control`` SDK model (not bedrock-agentcore-control). + + Required fields per the model: registryArn, recordArn, recordId, name, + recordType, recordVersion, status, createdAt, updatedAt. + """ + rid = record_id or f"rec-{name[:8].ljust(8, '0')}" + return { + "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/{rid}", + "recordId": rid, + "name": name, + "description": f"Test agent {name}", + "recordType": "CUSTOM", + "recordVersion": "1", + "status": status, + "createdAt": datetime(2026, 10, 1, tzinfo=timezone.utc), + "updatedAt": datetime(2026, 10, 25, tzinfo=timezone.utc), + } + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + +@pytest.fixture +def marker_store(monkeypatch): + store = {} + + def fake_get(sid): + return dict(store.get(sid, {})) + + def fake_set(sid, **updates): + merged = store.setdefault(sid, {}) + merged.update(updates) + return dict(merged) + + monkeypatch.setattr(postfab, "get_postfab_marker", fake_get) + monkeypatch.setattr(postfab, "set_postfab_marker", fake_set) + return store + + +@pytest.fixture +def jobs_table(monkeypatch): + table = mock.MagicMock() + ddb = mock.MagicMock() + ddb.Table.return_value = table + monkeypatch.setattr(postfab, "dynamodb", ddb) + monkeypatch.setattr(postfab, "FABRICATION_JOBS_TABLE", "jobs-test") + return table + + +@pytest.fixture +def registry_stubber(monkeypatch): + """Provide a botocore Stubber for the registry client. + + Replaces the lazy singleton so every test gets a fresh client+stubber. + Uses ``agent-registry-control`` (not ``bedrock-agentcore-control``) to + match the production client — the newer SDK model exposes the + ``filters``-based ListRegistryRecords shape the live service expects. + """ + client = boto3.client( + "agent-registry-control", region_name="ap-southeast-2", + ) + monkeypatch.setattr(postfab, "_registry_client", client) + monkeypatch.setattr(postfab, "REGISTRY_ID", "reg-test-1") + stubber = Stubber(client) + stubber.activate() + yield stubber + stubber.deactivate() + + +@pytest.fixture(autouse=True) +def _bind_session(): + """Bind a request context for every test (org-scoped lookups need it).""" + request_context.bind_request({ + "session_id": "sess-1", + "sessionAttributes": {"userId": "u1", "org": "org-1"}, + }) + + +def _stub_list(stubber, agent_name, registry_status): + """Stub a single ListRegistryRecords call returning one record. + + expected_params uses the ``filters`` shape — ListRegistryRecords has NO + top-level ``name`` parameter; the caller must pass + ``filters=[{name:'recordType',values:['CUSTOM']},{name:'name',values:[agent_name]}]``. + Mirrors arbiter/fabricator/index.py::_find_existing_record_id (source of truth). + """ + stubber.add_response( + "list_registry_records", + { + "registryRecords": [_registry_record(agent_name, registry_status)], + }, + expected_params={ + "registryId": "reg-test-1", + "filters": [ + {"name": "recordType", "values": ["CUSTOM"]}, + {"name": "name", "values": [agent_name]}, + ], + "maxResults": 10, + }, + ) + + +def _run_status(jobs_table, items, marker_store): + jobs_table.query.return_value = {"Items": items} + return json.loads(postfab.check_fabrication_status()) + + +# =========================================================================== +# Live registry status: APPROVED +# =========================================================================== + +class TestLiveStatusApproved: + """Agent was pending_approval at fabrication time but has since been + approved on the Approvals page — live status should say 'built and approved'.""" + + def test_approved_overrides_stale_pending_snapshot( + self, jobs_table, marker_store, registry_stubber, + ): + # 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") + + 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["AgentA"] == "built and approved" + # Should NOT be in pending_approval list + assert result["pending_approval"] == [] + # Should offer activation + labels = [a["label"] for a in result["actions"]] + assert any("Activate" in label for label in labels) + + def test_approved_agent_not_gated_on_activation( + self, jobs_table, marker_store, registry_stubber, monkeypatch, + ): + """An agent whose registry record is now APPROVED should not be + blocked by the activation gate.""" + marker_store["sess-1"] = {"stage": "built"} + items = [_job("AgentA", "COMPLETED", registry_approval_status="pending_approval")] + + # The activation gate queries the jobs table with pagination + jobs_table.query.return_value = {"Items": items} + + # Stub for the activation gate's own live resolution + _stub_list(registry_stubber, "AgentA", "APPROVED") + + execute = mock.MagicMock() + execute.return_value = {"intakeActivateProjectAgents": { + "activated": ["AgentA"], "failed": [], "alreadyActive": [], + "matchedBy": "sessionId", + }} + monkeypatch.setattr(postfab.appsync_client, "execute", execute) + monkeypatch.setattr(postfab, "_propose_app_name", lambda sid: "Test App") + + result = json.loads(postfab.activate_agents()) + + assert result["ok"] is True + execute.assert_called_once() + + +# =========================================================================== +# Live registry status: PENDING_APPROVAL +# =========================================================================== + +class TestLiveStatusPendingApproval: + """Agent is genuinely still pending approval in the registry.""" + + 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") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "awaiting approval" in states["AgentA"] + assert result["pending_approval"] == ["AgentA"] + + +# =========================================================================== +# Live registry status: DRAFT +# =========================================================================== + +class TestLiveStatusDraft: + """Agent record was never submitted for approval — still in DRAFT.""" + + def test_draft_shows_not_submitted( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "COMPLETED")] + _stub_list(registry_stubber, "AgentA", "DRAFT") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "not yet submitted" in states["AgentA"] + assert result["draft_agents"] == ["AgentA"] + # Should NOT offer activation — draft agents can't be activated + labels = [a["label"] for a in result["actions"]] + assert not any("Activate" in label for label in labels) + + def test_draft_complete_section_mentions_registry( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "COMPLETED")] + _stub_list(registry_stubber, "AgentA", "DRAFT") + + result = _run_status(jobs_table, items, marker_store) + + assert "not been submitted" in result["summary"] + assert "registry" in result["summary"].lower() + + +# =========================================================================== +# Live registry status: REJECTED +# =========================================================================== + +class TestLiveStatusRejected: + """Agent's approval was rejected.""" + + def test_rejected_shows_clear_message( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "COMPLETED")] + _stub_list(registry_stubber, "AgentA", "REJECTED") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "rejected" in states["AgentA"] + + +# =========================================================================== +# Live registry status: DEPRECATED +# =========================================================================== + +class TestLiveStatusDeprecated: + """Agent's record has been deprecated.""" + + def test_deprecated_shows_clear_message( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "COMPLETED")] + _stub_list(registry_stubber, "AgentA", "DEPRECATED") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert "deprecated" in states["AgentA"] + + +# =========================================================================== +# Fallback: registry unreachable +# =========================================================================== + +class TestRegistryFallback: + """When the registry is unreachable, fall back to job row snapshot.""" + + def test_registry_error_falls_back_to_snapshot( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "COMPLETED", registry_approval_status="pending_approval")] + # Stub an error + registry_stubber.add_client_error( + "list_registry_records", + service_error_code="InternalServerException", + ) + + result = _run_status(jobs_table, items, marker_store) + + # Should fall back to snapshot: pending_approval + states = {a["name"]: a["state"] for a in result["agents"]} + assert "awaiting approval" in states["AgentA"] + assert result["pending_approval"] == ["AgentA"] + + def test_registry_unset_falls_back_to_snapshot( + self, jobs_table, marker_store, monkeypatch, + ): + monkeypatch.setattr(postfab, "REGISTRY_ID", "") + items = [_job("AgentA", "COMPLETED")] + + result = _run_status(jobs_table, items, marker_store) + + # No REGISTRY_ID -> assume auto-approved (legacy behavior) + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["AgentA"] == "built and approved" + + +# =========================================================================== +# Mixed statuses +# =========================================================================== + +class TestMixedLiveStatuses: + """Multiple agents with different live registry statuses.""" + + def test_one_approved_one_pending( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job("AgentA", "COMPLETED", registry_approval_status="pending_approval"), + _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") + + result = _run_status(jobs_table, items, marker_store) + + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["AgentA"] == "built and approved" + assert "awaiting approval" in states["AgentB"] + assert result["pending_approval"] == ["AgentB"] + + def test_approved_and_failed_sibling( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job("AgentA", "COMPLETED"), + _job("AgentB", "FAILED"), + ] + _stub_list(registry_stubber, "AgentA", "APPROVED") + + result = _run_status(jobs_table, items, marker_store) + + assert result["status"] == "partial" + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["AgentA"] == "built and approved" + + def test_draft_and_approved_together( + self, jobs_table, marker_store, registry_stubber, + ): + items = [ + _job("AgentA", "COMPLETED"), + _job("AgentB", "COMPLETED"), + ] + _stub_list(registry_stubber, "AgentA", "DRAFT") + _stub_list(registry_stubber, "AgentB", "APPROVED") + + result = _run_status(jobs_table, items, marker_store) + + assert result["status"] == "complete" + assert result["draft_agents"] == ["AgentA"] + assert result["pending_approval"] == [] + # Should not offer activation because AgentA is still DRAFT + assert "not been submitted" in result["summary"] + + +# =========================================================================== +# Bounded lookups +# =========================================================================== + +class TestBoundedLookups: + """Registry lookups are bounded to _MAX_REGISTRY_LOOKUPS (5).""" + + def test_only_first_five_agents_resolved( + self, jobs_table, marker_store, registry_stubber, + ): + names = [f"Agent{i}" for i in range(7)] + items = [_job(n, "COMPLETED") for n in names] + # Stub only the first 5 + for n in names[:5]: + _stub_list(registry_stubber, n, "APPROVED") + + result = _run_status(jobs_table, items, marker_store) + + # First 5 should have live status; last 2 fall back to snapshot + states = {a["name"]: a["state"] for a in result["agents"]} + for n in names[:5]: + assert states[n] == "built and approved" + # Last 2 get snapshot fallback (no registryApprovalStatus = auto-approved) + for n in names[5:]: + assert states[n] == "built and approved" + + +# =========================================================================== +# No raw enums +# =========================================================================== + +class TestNoRawEnums: + """Raw registry enums must never reach the conversation.""" + + 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") + + result = _run_status(jobs_table, items, marker_store) + + blob = json.dumps(result) + for raw in ("PENDING_APPROVAL", "COMPLETED", "FAILED", "PROCESSING", + "PENDING", "DRAFT", "REJECTED", "DEPRECATED"): + assert raw not in blob, f"raw enum {raw} leaked into tool output" + + 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") + + result = _run_status(jobs_table, items, marker_store) + + blob = json.dumps(result) + for raw in ("PENDING_APPROVAL", "COMPLETED", "FAILED", "PROCESSING", + "PENDING", "DRAFT", "REJECTED", "DEPRECATED"): + assert raw not in blob, f"raw enum {raw} leaked into tool output" + + 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") + + result = _run_status(jobs_table, items, marker_store) + + blob = json.dumps(result) + for raw in ("PENDING_APPROVAL", "COMPLETED", "FAILED", "PROCESSING", + "PENDING", "DRAFT", "REJECTED", "DEPRECATED"): + assert raw not in blob, f"raw enum {raw} leaked into tool output" + + +# =========================================================================== +# FAILED jobs unchanged +# =========================================================================== + +class TestFailedJobsUnchanged: + """FAILED jobs should NOT trigger registry lookups.""" + + def test_failed_job_not_queried( + self, jobs_table, marker_store, registry_stubber, + ): + items = [_job("AgentA", "FAILED")] + # No stubs added — if a registry call is made it will raise + + result = _run_status(jobs_table, items, marker_store) + + assert result["status"] == "all_failed" + states = {a["name"]: a["state"] for a in result["agents"]} + assert states["AgentA"] == "didn't finish" + + +# =========================================================================== +# ParamValidator: prove the old top-level-name shape is rejected +# =========================================================================== + +class TestParamValidatorRejectsTopLevelName: + """Validate the built kwargs against the REAL botocore service model. + + botocore.validate.ParamValidator.validate() runs the same validation the + SDK would run before making a wire call — it checks the kwargs against + the service model's input shape. This proves that the + ``agent-registry-control`` model (which matches the LIVE service) does + NOT accept a top-level ``name`` kwarg — only ``filters``. + + The older ``bedrock-agentcore-control`` model DID have ``name`` as a + query-string parameter, but the live service rejects it (the exact error + seen in production: 'Unknown parameter in input: name, must be one of: + registryId, maxResults, nextToken, filters'). + """ + + @staticmethod + def _get_operation_model(): + """Return the botocore OperationModel for ListRegistryRecords on the + agent-registry-control model (matches the LIVE service).""" + client = boto3.client( + "agent-registry-control", region_name="ap-southeast-2", + ) + service_model = client._service_model + return service_model.operation_model("ListRegistryRecords") + + def test_old_top_level_name_kwarg_rejected(self): + """The OLD shape (name=agent_name as top-level kwarg) MUST fail + validation on the agent-registry-control model — this is the exact + bug that caused the live error.""" + from botocore.validate import ParamValidator + + op_model = self._get_operation_model() + old_kwargs = { + "registryId": "reg-test-1", + "name": "arbiter-agent", + "maxResults": 10, + } + report = ParamValidator().validate(old_kwargs, op_model.input_shape) + assert report.has_errors(), ( + "ParamValidator should reject top-level 'name' kwarg on the " + "agent-registry-control model but it passed" + ) + # The report should mention 'name' as unknown + errors_str = report.generate_report() + assert "name" in errors_str.lower() + + def test_new_filters_shape_accepted(self): + """The NEW shape (filters-based) MUST pass validation — this is the + correct API shape mirrored from _find_existing_record_id.""" + from botocore.validate import ParamValidator + + op_model = self._get_operation_model() + new_kwargs = { + "registryId": "reg-test-1", + "filters": [ + {"name": "recordType", "values": ["CUSTOM"]}, + {"name": "name", "values": ["arbiter-agent"]}, + ], + "maxResults": 10, + } + report = ParamValidator().validate(new_kwargs, op_model.input_shape) + assert not report.has_errors(), ( + f"ParamValidator should accept filters-based shape but got errors: " + f"{report.generate_report()}" + ) diff --git a/service/agent_intake_single/tools/postfab.py b/service/agent_intake_single/tools/postfab.py index 61b8356..62cd44f 100644 --- a/service/agent_intake_single/tools/postfab.py +++ b/service/agent_intake_single/tools/postfab.py @@ -43,9 +43,36 @@ FABRICATION_JOBS_TABLE = os.environ.get("FABRICATION_JOBS_TABLE", "") PROJECTS_TABLE = os.environ.get("PROJECTS_TABLE", "") CONVERSATIONS_TABLE = os.environ.get("CONVERSATIONS_TABLE", "") +REGISTRY_ID = os.environ.get("REGISTRY_ID", "") dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION) +# Lazy registry client — built on first use so module import stays cheap. +_registry_client = None + + +def _get_registry_client(): + """Lazy boto3 client for agent-registry-control Registry APIs. + + Uses ``agent-registry-control`` (not ``bedrock-agentcore-control``) to + match arbiter/fabricator/index.py — the live service accepts the + ``filters``-based ListRegistryRecords shape exposed by this newer SDK + model and rejects the ``name`` query-string parameter from the older + ``bedrock-agentcore-control`` model. + """ + global _registry_client + if _registry_client is None: + _registry_client = boto3.client( + "agent-registry-control", region_name=AWS_REGION, + ) + return _registry_client + + +def _reset_registry_client_for_test() -> None: + """Test-only hook — forces the next call to rebuild the cached client.""" + global _registry_client + _registry_client = None + # Plain-language translations for fabrication job statuses — raw enums must # never reach the conversation. _STATUS_PLAIN = { @@ -61,20 +88,120 @@ # AgentCore Registry's SubmitRegistryRecordForApproval returns # PENDING_APPROVAL (manual-review mode). Absent or any other value means # the record was auto-approved. +# +# LIVE STATUS RESOLUTION (fix/intake-build-status-from-registry): +# The job row's registryApprovalStatus is a WRITE-ONCE snapshot taken at +# fabrication time. After a human approves (or rejects) the record on the +# Approvals page, the snapshot is stale. For COMPLETED jobs we now resolve +# the agent's registry record by name and derive state from the LIVE status. +# The job row value is only used as a fallback when the registry is +# unreachable or REGISTRY_ID is unset. _APPROVAL_PLAIN = { "pending_approval": "built, awaiting approval on the Approvals page", "approved": "built and approved", } -def _plain_state(item: dict) -> str: - """Human-readable state for a single job row, approval-aware. +# Registry status -> human-readable state for COMPLETED jobs. +_REGISTRY_STATUS_PLAIN: dict[str, str] = { + "DRAFT": "built, not yet submitted for approval", + "PENDING_APPROVAL": "awaiting approval on the Approvals page", + "APPROVED": "built and approved", + "REJECTED": "built, but the approval was rejected", + "DEPRECATED": "built, but has been deprecated", +} + +# Max number of COMPLETED agents to resolve via registry reads per +# check_fabrication_status call — keeps latency bounded. +_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. - COMPLETED rows with ``registryApprovalStatus='pending_approval'`` read as - 'built, awaiting approval on the Approvals page'; COMPLETED without that - flag (auto-approved or flag absent) read as 'built and approved'. + 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: + 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). + response = client.list_registry_records( + registryId=REGISTRY_ID, + filters=[ + {"name": "recordType", "values": ["CUSTOM"]}, + {"name": "name", "values": [agent_name]}, + ], + maxResults=10, + ) + if not isinstance(response, dict): + return None + for summary in response.get("registryRecords", []): + if not isinstance(summary, dict): + continue + if summary.get("name") == agent_name: + return summary.get("status") + return None + except Exception as e: + logger.warning( + "registry status lookup failed agent=%s: %s", + agent_name, type(e).__name__, + ) + return None + + +def _resolve_live_statuses( + items: list[dict], +) -> dict[str, str | None]: + """Resolve live registry statuses for COMPLETED job rows. + + Returns a dict mapping agentName -> registry status string (or None + 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. + """ + result: dict[str, str | None] = {} + # Derive org_id from the current session for org-scoped lookups. + org_id = "" + try: + org_id = request_context.current_session().org_id + except (RuntimeError, AttributeError): + pass + lookups = 0 + for item in items: + name = item.get("agentName") or item.get("agentUseId") + if not name: + continue + if item.get("status") != "COMPLETED": + continue + if lookups >= _MAX_REGISTRY_LOOKUPS: + result[name] = None + continue + lookups += 1 + result[name] = _resolve_registry_status(name, org_id) + return result + + +def _plain_state(item: dict, live_status: str | None = None) -> str: + """Human-readable state for a single job row, live-registry-aware. + + When ``live_status`` is provided (from a registry lookup), it takes + precedence over the job row's write-once ``registryApprovalStatus``. + Falls back to the snapshot when the registry is unreachable. """ status = item.get("status") if status == "COMPLETED": + if live_status is not None: + return _REGISTRY_STATUS_PLAIN.get( + live_status, + f"built (registry status: {live_status.lower()})", + ) + # Fallback to job row snapshot. approval = item.get("registryApprovalStatus") if approval == "pending_approval": return _APPROVAL_PLAIN["pending_approval"] @@ -82,10 +209,28 @@ def _plain_state(item: dict) -> str: return _STATUS_PLAIN.get(status, "in an unknown state") -def _is_pending_approval(item: dict) -> bool: - """True when a COMPLETED job still needs registry approval.""" - return (item.get("status") == "COMPLETED" - and item.get("registryApprovalStatus") == "pending_approval") +def _is_pending_approval( + item: dict, live_status: str | None = None, +) -> bool: + """True when a COMPLETED job still needs registry approval. + + Checks the LIVE registry status first; falls back to the job row + snapshot when the registry is unreachable. + """ + if item.get("status") != "COMPLETED": + return False + if live_status is not None: + return live_status == "PENDING_APPROVAL" + return item.get("registryApprovalStatus") == "pending_approval" + + +def _is_draft(item: dict, live_status: str | None = None) -> bool: + """True when a COMPLETED job's registry record is still in DRAFT.""" + if item.get("status") != "COMPLETED": + return False + if live_status is not None: + return live_status == "DRAFT" + return False _STAGE_ORDER = [ @@ -443,9 +588,21 @@ def check_fabrication_status() -> str: def _name(i: dict) -> str: return i.get("agentName") or i.get("agentUseId") + # Resolve LIVE registry statuses for COMPLETED agents (bounded, sequential). + live_statuses = _resolve_live_statuses(items) + + def _live(i: dict) -> str | None: + """Live registry status for this item, or None if not resolved.""" + return live_statuses.get(_name(i)) + succeeded = [_name(i) for i in items if i.get("status") == "COMPLETED"] failed = [_name(i) for i in items if i.get("status") == "FAILED"] - pending_approval = [_name(i) for i in items if _is_pending_approval(i)] + pending_approval = [ + _name(i) for i in items if _is_pending_approval(i, _live(i)) + ] + draft_agents = [ + _name(i) for i in items if _is_draft(i, _live(i)) + ] # Non-terminal rows split by updatedAt age: a stale row is orphaned — its # writer can never re-stamp it (see STALE_ACTIVE_SECONDS in # tools/fabricate.py for the derivation) — so it gates like a failure @@ -461,7 +618,7 @@ def _name(i: dict) -> str: {"name": _name(i), "state": "stalled" if i.get("status") not in _TERMINAL and _name(i) in stale_set - else _plain_state(i)} + else _plain_state(i, _live(i))} for i in items ] counts = { @@ -495,6 +652,7 @@ def _name(i: dict) -> str: "building": building, "stalled": stale, "all_terminal": all_terminal, "all_succeeded": all_succeeded, "any_failed": any_failed, "pending_approval": pending_approval, + "draft_agents": draft_agents, } if status == "none": @@ -549,6 +707,27 @@ def _name(i: dict) -> str: ) if status == "complete": + if draft_agents: + # DRAFT = never submitted for approval. The intake tools do not + # have a submit-for-approval action today — the user must submit + # from the registry UI or an operator must approve directly. + draft_summary = ( + f"All {total} agents built successfully. " + f"{_join_names(draft_agents)} " + + ("has" if len(draft_agents) == 1 else "have") + + " not been submitted for approval yet — you can submit " + + ("it" if len(draft_agents) == 1 else "them") + + " from the agent registry." + ) + return _result( + True, "complete", + draft_summary, + "Submit the agents for approval in the registry, then come " + "back and check status again.", + [_action("Check again", "Check the build status again"), + _action("Not now", "Not now")], + **extra, + ) if pending_approval: pa_summary = ( f"All {total} agents built successfully. " @@ -585,19 +764,26 @@ def _name(i: dict) -> str: ) if stale: bits.append(_stalled_sentence(stale)) + if draft_agents: + bits.append( + f"{_join_names(draft_agents)} " + + ("has" if len(draft_agents) == 1 else "have") + + " not been submitted for approval yet." + ) if pending_approval: bits.append( f"{_join_names(pending_approval)} " + ("is" if len(pending_approval) == 1 else "are") + " awaiting approval on the Approvals page." ) - # Activatable = completed agents that are NOT pending approval. - activatable = [n for n in succeeded if n not in pending_approval] - if pending_approval and not activatable: - # All completed agents are pending — can't activate yet. + # Activatable = completed agents that are NOT pending approval and NOT draft. + not_activatable = set(pending_approval) | set(draft_agents) + activatable = [n for n in succeeded if n not in not_activatable] + if not_activatable and not activatable: + # All completed agents are pending/draft — can't activate yet. actions = [_action("Check again", "Check the build status again")] else: - activate_count = len(activatable) if pending_approval else len(succeeded) + activate_count = len(activatable) if not_activatable else len(succeeded) actions = [_action(f"Activate {activate_count} agents", "Yes, activate the agents that are ready")] retryable = failed + stale @@ -607,11 +793,11 @@ def _name(i: dict) -> str: actions.append(_action(f"Look at {_join_names(failed[:1])} first", "Let's look at the failed agent first")) actions.append(_action("Not now", "Not now")) - activate_count = len(activatable) if pending_approval else len(succeeded) + activate_count = len(activatable) if not_activatable else len(succeeded) consent_q = ( "Once approved, I can activate them. Check the Approvals page, " "then come back and check status again." - if pending_approval and not activatable + if not_activatable and not activatable else f"Want me to activate the {activate_count} that are ready?" ) return _result( @@ -676,9 +862,17 @@ def activate_agents() -> str: if not resp.get("LastEvaluatedKey"): break kwargs["ExclusiveStartKey"] = resp["LastEvaluatedKey"] + # Resolve live registry statuses for the gate check (bounded). + gate_live = _resolve_live_statuses(pa_items) pa_names = [ i.get("agentName") or i.get("agentUseId") - for i in pa_items if _is_pending_approval(i) + for i in pa_items + if _is_pending_approval( + i, + gate_live.get( + i.get("agentName") or i.get("agentUseId"), + ), + ) ] if pa_names: return _result(