fix(a2a): stabilize permission checkpoint identity

scripts/a2a/e2e/permission_wait/permission_wait_fixture_server.py

@@ -491,12 +491,30 @@ def main() -> int:
     os.environ["IAC_CODE_CONFIG_DIR"] = str(config_dir)
     os.environ["IAC_CODE_MODE"] = args.mode
     os.environ["IACCODE_A2A_ALLOWED_CWDS"] = str(workspace)
+    # Cloud-shaped permission fixtures must exercise the durable identity
+    # contract without making a real STS request.  Keep one stable caller
+    # across both fixture-server generations while using explicitly fake
+    # credentials for the request-local credential plumbing.
+    os.environ["ALIBABA_CLOUD_ACCESS_KEY_ID"] = "permission-wait-fixture-ak"
+    os.environ["ALIBABA_CLOUD_ACCESS_KEY_SECRET"] = "permission-wait-fixture-secret"
+    os.environ["ALIBABA_CLOUD_REGION_ID"] = "cn-hangzhou"
 
     import uvicorn
 
     from iac_code.a2a import executor as executor_module
     from iac_code.a2a import pipeline_executor as pipeline_executor_module
     from iac_code.a2a.app import create_app
+    from iac_code.services.providers.aliyun_identity import AliyunCallerIdentity, AliyunCallerIdentityResolver
+
+    async def resolve_fixture_identity(
+        self: AliyunCallerIdentityResolver,
+        credential: Any,
+        region_id: str,
+    ) -> AliyunCallerIdentity:
+        del self, credential, region_id
+        return AliyunCallerIdentity(kind="ram_role", account_id="1000000000000001", subject_id="fixture-role-id")
+
+    AliyunCallerIdentityResolver.resolve = resolve_fixture_identity
 
     executor_module.create_agent_runtime = lambda options: _create_fixture_runtime(
         options,

src/iac_code/a2a/executor.py

@@ -31,6 +31,7 @@
 )
 from iac_code.a2a.exposure import normalize_a2a_exposure_types
 from iac_code.a2a.input_required import (
+    PermissionIdentityValidationError,
     PermissionInputRegistry,
     PermissionResponse,
     backup_permission_wait_checkpoint,
@@ -110,11 +111,12 @@
     PermissionWaitCheckpointStore,
     RecoveredPermissionAuditBoundary,
     canonical_digest,
-    permission_execution_identity,
     recover_permission_audit_boundary,
+    resolve_permission_execution_identity,
 )
 from iac_code.services.permissions.audit import emit_permission_boundary_audit
 from iac_code.services.providers.aliyun import DEFAULT_REGION, AliyunCredential, AliyunCredentials
+from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityUnavailableError
 from iac_code.services.session_backup import (
     BackupReason,
     SessionBackupBlocked,
@@ -1245,10 +1247,23 @@ async def _execute(self, context: RequestContext, event_queue: EventQueue, *, co
         task_id = requested_task_id or "task-" + uuid.uuid4().hex[:12]
         permission_response = parse_permission_response(getattr(context, "message", None))
         if permission_response is not None:
+            response_metadata = getattr(context, "metadata", None) or getattr(
+                getattr(context, "message", None), "metadata", None
+            )
+            response_credential = self._resolve_aliyun_credential(response_metadata)
             pending = None
             try:
                 pending = await self._permission_input_registry.pending_for_response(permission_response)
-                approved = await self._permission_input_registry.answer(permission_response)
+                with a2a_request_context(aliyun_credential=response_credential):
+                    approved = await self._permission_input_registry.answer(permission_response)
+            except PermissionIdentityValidationError as exc:
+                await self._publish_permission_identity_error(
+                    event_queue,
+                    response=permission_response,
+                    code=exc.code,
+                    retryable=exc.retryable,
+                )
+                return
             except InvalidParamsError:
                 if pending is not None:
                     await self._permission_input_registry.complete(pending)
@@ -2251,15 +2266,46 @@ def make_pipeline_executor() -> IacCodeA2APipelineExecutor:
             except (OSError, RuntimeError, TypeError, ValueError) as exc:
                 raise InvalidParamsError(f"permission_resume_invalid: {exc}") from exc
 
-            permission_audit = getattr(audit_event.permission_result, "audit", None)
-            with a2a_request_context(aliyun_credential=aliyun_credential):
-                principal_ref, region = permission_execution_identity(
-                    tool_name=audit_event.tool_name,
-                    tool_input=audit_event.tool_input,
-                    permission_audit=permission_audit,
-                )
-            if principal_ref != record.get("principalRef") or region != record.get("region"):
-                raise InvalidParamsError("permission_resume_invalid: cloud execution identity changed.")
+            if expected_value == "allow_once":
+                if record.get("identitySchemaVersion") != 1:
+                    await self._publish_permission_identity_error(
+                        event_queue,
+                        response=response,
+                        code="legacy_permission_identity",
+                        retryable=False,
+                        session_id=context_record.session_id,
+                    )
+                    return True
+                permission_audit = getattr(audit_event.permission_result, "audit", None)
+                try:
+                    with a2a_request_context(aliyun_credential=aliyun_credential):
+                        identity = await resolve_permission_execution_identity(
+                            tool_name=audit_event.tool_name,
+                            tool_input=audit_event.tool_input,
+                            permission_audit=permission_audit,
+                        )
+                except AliyunCallerIdentityUnavailableError as exc:
+                    await self._publish_permission_identity_error(
+                        event_queue,
+                        response=response,
+                        code=exc.reason,
+                        retryable=exc.retryable,
+                        session_id=context_record.session_id,
+                    )
+                    return True
+                if (
+                    identity.principal_ref != record.get("principalRef")
+                    or identity.principal_kind != record.get("principalKind")
+                    or identity.region != record.get("region")
+                ):
+                    await self._publish_permission_identity_error(
+                        event_queue,
+                        response=response,
+                        code="cloud_execution_identity_changed",
+                        retryable=False,
+                        session_id=context_record.session_id,
+                    )
+                    return True
 
         record = store.reconcile_deadline(
             boundary_id,
@@ -2595,6 +2641,34 @@ async def _publish_permission_recovery_ack(
             session_id=session_id,
         )
 
+    async def _publish_permission_identity_error(
+        self,
+        event_queue: EventQueue,
+        *,
+        response: PermissionResponse,
+        code: str,
+        retryable: bool,
+        session_id: str | None = None,
+    ) -> None:
+        await self._publish_status(
+            event_queue,
+            task_id=response.task_id,
+            context_id=response.context_id,
+            state=TaskState.TASK_STATE_INPUT_REQUIRED,
+            text=_("A temporary error occurred. Please retry."),
+            metadata={
+                "iac_code": {
+                    "permissionIdentityError": {
+                        "code": code,
+                        "retryable": retryable,
+                        "inputId": response.input_id,
+                        "toolUseId": response.tool_use_id,
+                    }
+                }
+            },
+            session_id=session_id,
+        )
+
     async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
         task_id = context.task_id
         context_id = context.context_id or "unknown"
@@ -3354,6 +3428,8 @@ def _is_normal_handoff(handoff: dict[str, Any]) -> bool:
 
 
 def _is_retryable_executor_error(exc: Exception) -> bool:
+    if isinstance(exc, AliyunCallerIdentityUnavailableError):
+        return exc.retryable
     return isinstance(
         exc,
         (

src/iac_code/a2a/input_required.py

@@ -21,14 +21,15 @@
     PermissionWaitPolicy,
     build_permission_checkpoint,
     canonicalize_permission_continuation_frame,
-    permission_execution_identity,
+    resolve_permission_execution_identity,
 )
 from iac_code.services.permissions.audit import (
     build_display_tool_input,
     build_prompt_tool_input,
     emit_permission_boundary_audit,
     sanitize_prompt_text,
 )
+from iac_code.services.providers.aliyun_identity import AliyunCallerIdentityUnavailableError
 from iac_code.types.stream_events import PermissionRequestEvent
 
 PERMISSION_SCHEMA_VERSION = 1
@@ -37,6 +38,16 @@
 _SAFE_SUMMARY_MAX_CHARS = 1200
 _DISPLAY_FIELD_MAX_CHARS = 500
 
+
+class PermissionIdentityValidationError(InvalidParamsError):
+    """A permission decision could not safely use the current cloud identity."""
+
+    def __init__(self, code: str, *, retryable: bool) -> None:
+        super().__init__(f"permission_resume_invalid: {code}")
+        self.code = code
+        self.retryable = retryable
+
+
 _ROS_TEMPLATE_API_ACTIONS = {
     "ros_validate_template": "ValidateTemplate",
     "ros_get_template_parameter_constraints": "GetTemplateParameterConstraints",
@@ -878,8 +889,12 @@ def fenced_backup(_cwd: str, _session_id: str, **kwargs: Any) -> Any:
         )
     except ValueError as exc:
         raise SessionBackupBlocked("Permission checkpoint changed during critical backup.") from exc
-    if getattr(result, "enabled", False) and not getattr(result, "shared_committed", False):
-        raise SessionBackupBlocked("Critical permission backup did not reach the shared target.")
+    if (
+        getattr(result, "enabled", False)
+        and not getattr(result, "shared_committed", False)
+        and not getattr(result, "staged_committed", False)
+    ):
+        raise SessionBackupBlocked("Critical permission backup did not reach a durable target.")
     return result
 
 
@@ -932,8 +947,17 @@ async def open_durable_boundary(
             frame = canonicalize_permission_continuation_frame(source_frame, audit_context=audit_context)
         except ValueError as exc:
             raise RuntimeError(f"permission_resume_invalid: {exc}") from exc
-        principal_ref = audit_context.get("principal_ref")
-        region = audit_context.get("region")
+        permission_audit = getattr(pending.request.permission_result, "audit", None)
+        execution_identity = await resolve_permission_execution_identity(
+            tool_name=pending.request.tool_name,
+            tool_input=pending.request.tool_input,
+            permission_audit=permission_audit,
+        )
+        audit_context.update(
+            principal_ref=execution_identity.principal_ref,
+            principal_kind=execution_identity.principal_kind,
+            region=execution_identity.region,
+        )
         record = build_permission_checkpoint(
             session_id=session_id,
             task_id=pending.task_id,
@@ -945,8 +969,9 @@ async def open_durable_boundary(
             permission_class="pipeline" if permission_class == "pipeline" else "normal",
             continuation_frame=frame,
             policy=coordinator.policy,
-            principal_ref=principal_ref if isinstance(principal_ref, str) else None,
-            region=region if isinstance(region, str) else None,
+            principal_ref=execution_identity.principal_ref,
+            principal_kind=execution_identity.principal_kind,
+            region=execution_identity.region,
             pipeline_coordinates=pipeline_coordinates,
         )
         previous_boundary_id = frame.get("previousBoundaryId")
@@ -1112,7 +1137,8 @@ async def answer(self, response: PermissionResponse) -> bool:
 
         coordinator = self._permission_wait_coordinator
         if coordinator is not None and pending.boundary_id is not None:
-            self._validate_live_execution_identity(pending)
+            if response.decision == "allow_once":
+                await self._validate_live_execution_identity(pending)
 
             def audit_new_claim(value: str) -> bool:
                 return emit_permission_boundary_audit(
@@ -1160,7 +1186,7 @@ def audit_new_claim(value: str) -> bool:
             return approved
 
     @staticmethod
-    def _validate_live_execution_identity(pending: PendingPermission) -> None:
+    async def _validate_live_execution_identity(pending: PendingPermission) -> None:
         store = pending.checkpoint_store
         boundary_id = pending.boundary_id
         if store is None or boundary_id is None:
@@ -1169,13 +1195,22 @@ def _validate_live_execution_identity(pending: PendingPermission) -> None:
         if record is None:
             raise InvalidParamsError("permission_resume_invalid: permission checkpoint is unavailable.")
         permission_audit = getattr(pending.request.permission_result, "audit", None)
-        principal_ref, region = permission_execution_identity(
-            tool_name=pending.request.tool_name,
-            tool_input=pending.request.tool_input,
-            permission_audit=permission_audit,
-        )
-        if principal_ref != record.get("principalRef") or region != record.get("region"):
-            raise InvalidParamsError("permission_resume_invalid: cloud execution identity changed.")
+        if record.get("identitySchemaVersion") != 1:
+            raise PermissionIdentityValidationError("legacy_permission_identity", retryable=False)
+        try:
+            identity = await resolve_permission_execution_identity(
+                tool_name=pending.request.tool_name,
+                tool_input=pending.request.tool_input,
+                permission_audit=permission_audit,
+            )
+        except AliyunCallerIdentityUnavailableError as exc:
+            raise PermissionIdentityValidationError(exc.reason, retryable=exc.retryable) from exc
+        if (
+            identity.principal_ref != record.get("principalRef")
+            or identity.principal_kind != record.get("principalKind")
+            or identity.region != record.get("region")
+        ):
+            raise PermissionIdentityValidationError("cloud_execution_identity_changed", retryable=False)
 
     async def _backup_claim_before_delivery(self, pending: PendingPermission, record: dict[str, Any]) -> None:
         if pending.before_claim_backup is not None:
@@ -1413,6 +1448,7 @@ def _permission_input_id() -> str:
 __all__ = [
     "PERMISSION_QUERY_PREFIX",
     "PendingPermission",
+    "PermissionIdentityValidationError",
     "PermissionInputRegistry",
     "PermissionResolutionOwner",
     "PermissionResponse",

src/iac_code/a2a/runtime_overrides.py

@@ -9,6 +9,7 @@
 
 from iac_code.i18n import SUPPORTED_LANGUAGES, use_request_language
 from iac_code.providers.request_policy import ProviderRequestPolicy
+from iac_code.services.permission_wait import permission_execution_identity_cache_scope
 from iac_code.services.providers.aliyun import AliyunCredential, use_aliyun_credential
 from iac_code.services.telemetry import use_session_id, use_telemetry_channel, use_user_id
 
@@ -52,6 +53,7 @@ def a2a_request_context(
     telemetry_channel: str | None = None,
 ) -> Iterator[None]:
     with contextlib.ExitStack() as stack:
+        stack.enter_context(permission_execution_identity_cache_scope())
         if telemetry_channel:
             stack.enter_context(use_telemetry_channel(telemetry_channel))
         if preferred_language:

src/iac_code/agent/agent_loop.py

@@ -1257,15 +1257,18 @@ async def resume_permission_boundary(
             if request_index == current_index:
                 if permission is None:
                     raise ValueError("permission_resume_invalid: current tool is unavailable")
-                principal_ref, region = permission_execution_identity(
-                    tool_name=request.name,
-                    tool_input=request.input,
-                    permission_audit=getattr(permission, "audit", None),
-                )
-                if principal_ref != checkpoint.get("principalRef") or region != checkpoint.get("region"):
-                    raise ValueError("permission_resume_invalid: cloud execution identity changed")
                 state = "allow" if decision["value"] == "allow_once" else "deny"
                 source = "user"
+                principal_ref = None
+                region = None
+                if state == "allow":
+                    principal_ref, region = permission_execution_identity(
+                        tool_name=request.name,
+                        tool_input=request.input,
+                        permission_audit=getattr(permission, "audit", None),
+                    )
+                    if principal_ref != checkpoint.get("principalRef") or region != checkpoint.get("region"):
+                        raise ValueError("permission_resume_invalid: cloud execution identity changed")
                 if state == "deny":
                     recorded["deniedResult"] = _user_denied_tool_result()
                 if permission.behavior != "deny":
@@ -2198,6 +2201,15 @@ async def _run_streaming_inner(
                         approved = False
                         denial_source = "audit_failure"
                     if approved:
+                        # A2A durable permission publication may have resolved a
+                        # stable STS caller while this generator was suspended.
+                        # Re-read the request-local identity so continuation
+                        # decisions never retain a rotating temporary AccessKeyId.
+                        principal_ref, region = permission_execution_identity(
+                            tool_name=request.name,
+                            tool_input=request.input,
+                            permission_audit=permission.audit,
+                        )
                         allowed_requests.append(request)
                         continuation_decisions[request_index].update(
                             state="allow",